authentik.providers.oauth2.errors

  1"""OAuth errors"""
  2
  3from urllib.parse import quote, urlparse
  4
  5from django.http import HttpRequest, HttpResponse, HttpResponseRedirect
  6
  7from authentik.events.models import Event, EventAction
  8from authentik.lib.sentry import SentryIgnoredException
  9from authentik.lib.views import bad_request_message
 10from authentik.providers.oauth2.models import GrantTypes, RedirectURI
 11
 12
 13class OAuth2Error(SentryIgnoredException):
 14    """Base class for all OAuth2 Errors"""
 15
 16    error: str
 17    description: str
 18    cause: str | None = None
 19
 20    def create_dict(self, request: HttpRequest):
 21        """Return error as dict for JSON Rendering"""
 22        return {
 23            "error": self.error,
 24            "error_description": self.description,
 25            "request_id": request.request_id,
 26        }
 27
 28    def __repr__(self) -> str:
 29        return self.error
 30
 31    def to_event(self, message: str | None = None, **kwargs) -> Event:
 32        """Create configuration_error Event."""
 33        return Event.new(
 34            EventAction.CONFIGURATION_ERROR,
 35            message=message or self.description,
 36            cause=self.cause,
 37            error=self.error,
 38            **kwargs,
 39        )
 40
 41    def with_cause(self, cause: str):
 42        self.cause = cause
 43        return self
 44
 45
 46class RedirectUriError(OAuth2Error):
 47    """The request fails due to a missing, invalid, or mismatching
 48    redirection URI (redirect_uri)."""
 49
 50    error = "Redirect URI Error"
 51    description = (
 52        "The request fails due to a missing, invalid, or mismatching "
 53        "redirection URI (redirect_uri)."
 54    )
 55
 56    provided_uri: str
 57    allowed_uris: list[RedirectURI]
 58
 59    def __init__(self, provided_uri: str, allowed_uris: list[RedirectURI]) -> None:
 60        super().__init__()
 61        self.provided_uri = provided_uri
 62        self.allowed_uris = allowed_uris
 63
 64    def to_event(self, **kwargs) -> Event:
 65        return super().to_event(
 66            (
 67                f"Invalid redirect URI was used. Client used '{self.provided_uri}'. "
 68                f"Allowed redirect URIs are {','.join(self.allowed_uris)}"
 69            ),
 70            **kwargs,
 71        )
 72
 73
 74class ClientIdError(OAuth2Error):
 75    """The client identifier (client_id) is missing or invalid."""
 76
 77    error = "Client ID Error"
 78    description = "The client identifier (client_id) is missing or invalid."
 79
 80    client_id: str
 81
 82    def __init__(self, client_id: str) -> None:
 83        super().__init__()
 84        self.client_id = client_id
 85
 86    def to_event(self, **kwargs) -> Event:
 87        return super().to_event(f"Invalid client identifier: {self.client_id}.", **kwargs)
 88
 89
 90class UserAuthError(OAuth2Error):
 91    """
 92    Specific to the Resource Owner Password Credentials flow when
 93    the Resource Owners credentials are not valid.
 94    """
 95
 96    error = "access_denied"
 97    description = "The resource owner or authorization server denied the request."
 98
 99
100class TokenIntrospectionError(OAuth2Error):
101    """
102    Specific to the introspection endpoint. This error will be converted
103    to an "active: false" response, as per the spec.
104    See https://datatracker.ietf.org/doc/html/rfc7662
105    """
106
107
108class AuthorizeError(OAuth2Error):
109    """General Authorization Errors"""
110
111    errors = {
112        # OAuth2 errors.
113        # https://datatracker.ietf.org/doc/html/rfc6749#section-4.1.2.1
114        "invalid_request": "The request is otherwise malformed",
115        "unauthorized_client": (
116            "The client is not authorized to request an authorization code using this method"
117        ),
118        "access_denied": "The resource owner or authorization server denied the request",
119        "unsupported_response_type": (
120            "The authorization server does not support obtaining an authorization code "
121            "using this method"
122        ),
123        "invalid_scope": "The requested scope is invalid, unknown, or malformed",
124        "server_error": "The authorization server encountered an error",
125        "temporarily_unavailable": (
126            "The authorization server is currently unable to handle the request due to a "
127            "temporary overloading or maintenance of the server"
128        ),
129        # OpenID errors.
130        # http://openid.net/specs/openid-connect-core-1_0.html#AuthError
131        "interaction_required": (
132            "The Authorization Server requires End-User interaction of some form to proceed"
133        ),
134        "login_required": "The Authorization Server requires End-User authentication",
135        "account_selection_required": (
136            "The End-User is required to select a session at the Authorization Server"
137        ),
138        "consent_required": "The Authorization Server requires End-User consent",
139        "invalid_request_uri": (
140            "The request_uri in the Authorization Request returns an error or contains invalid data"
141        ),
142        "invalid_request_object": "The request parameter contains an invalid Request Object",
143        "request_not_supported": "The provider does not support use of the request parameter",
144        "request_uri_not_supported": (
145            "The provider does not support use of the request_uri parameter"
146        ),
147        "registration_not_supported": (
148            "The provider does not support use of the registration parameter"
149        ),
150    }
151
152    def __init__(
153        self,
154        redirect_uri: str,
155        error: str,
156        grant_type: str,
157        state: str,
158        description: str | None = None,
159    ):
160        super().__init__()
161        self.error = error
162        if description:
163            self.description = description
164        else:
165            self.description = self.errors[error]
166        self.redirect_uri = redirect_uri
167        self.grant_type = grant_type
168        self.state = state
169
170    def get_response(self, request: HttpRequest) -> HttpResponse:
171        """Wrapper around `self.create_uri()` that checks if the resulting URI is valid
172        (we might not have self.redirect_uri set), and returns a valid HTTP Response"""
173        uri = self.create_uri()
174        if urlparse(uri).scheme != "":
175            return HttpResponseRedirect(uri)
176        return bad_request_message(request, self.description, title=self.error)
177
178    def create_uri(self) -> str:
179        """Get a redirect URI with the error message"""
180        description = quote(str(self.description))
181
182        # See:
183        # http://openid.net/specs/openid-connect-core-1_0.html#ImplicitAuthError
184        fragment_or_query = (
185            "#" if self.grant_type in [GrantTypes.IMPLICIT, GrantTypes.HYBRID] else "?"
186        )
187
188        uri = (
189            f"{self.redirect_uri}{fragment_or_query}error="
190            f"{self.error}&error_description={description}"
191        )
192
193        # Add state if present.
194        uri = uri + (f"&state={self.state}" if self.state else "")
195
196        return uri
197
198
199class TokenError(OAuth2Error):
200    """
201    OAuth2 token endpoint errors.
202    https://datatracker.ietf.org/doc/html/rfc6749#section-5.2
203    """
204
205    errors = {
206        "invalid_request": "The request is otherwise malformed",
207        "invalid_client": (
208            "Client authentication failed (e.g., unknown client, no client authentication "
209            "included, or unsupported authentication method)"
210        ),
211        "invalid_grant": (
212            "The provided authorization grant or refresh token is invalid, expired, revoked, "
213            "does not match the redirection URI used in the authorization request, "
214            "or was issued to another client"
215        ),
216        "unauthorized_client": (
217            "The authenticated client is not authorized to use this authorization grant type"
218        ),
219        "unsupported_grant_type": (
220            "The authorization grant type is not supported by the authorization server"
221        ),
222        "invalid_scope": (
223            "The requested scope is invalid, unknown, malformed, or exceeds the scope "
224            "granted by the resource owner"
225        ),
226    }
227
228    def __init__(self, error):
229        super().__init__()
230        self.error = error
231        self.description = self.errors[error]
232
233
234class TokenRevocationError(OAuth2Error):
235    """
236    Specific to the revocation endpoint.
237    See https://datatracker.ietf.org/doc/html/rfc7662
238    """
239
240    errors = TokenError.errors | {
241        "unsupported_token_type": (
242            "The authorization server does not support the revocation of the presented "
243            "token type.  That is, the client tried to revoke an access token on a server not"
244            "supporting this feature."
245        )
246    }
247
248    def __init__(self, error: str):
249        super().__init__()
250        self.error = error
251        self.description = self.errors[error]
252
253
254class DeviceCodeError(TokenError):
255    """
256    Device-code flow errors
257    See https://datatracker.ietf.org/doc/html/rfc8628#section-3.2
258    Can also use codes form TokenError
259    """
260
261    errors = TokenError.errors | {
262        "authorization_pending": (
263            "The authorization request is still pending as the end user hasn't "
264            "yet completed the user-interaction steps"
265        ),
266        "access_denied": "The authorization request was denied.",
267        "expired_token": (
268            'The "device_code" has expired, and the device authorization '
269            "session has concluded.  The client MAY commence a new device "
270            "authorization request but SHOULD wait for user interaction before "
271            "restarting to avoid unnecessary polling."
272        ),
273        "slow_down": (
274            'A variant of "authorization_pending", the authorization request is'
275            "still pending and polling should continue, but the interval MUST"
276            "be increased by 5 seconds for this and all subsequent requests."
277        ),
278    }
279
280    def __init__(self, error: str):
281        super().__init__(error)
282        self.error = error
283        self.description = self.errors[error]
284
285
286class BearerTokenError(OAuth2Error):
287    """
288    OAuth2 errors.
289    https://datatracker.ietf.org/doc/html/rfc6750#section-3.1
290    """
291
292    errors = {
293        "invalid_request": ("The request is otherwise malformed", 400),
294        "invalid_token": (
295            (
296                "The access token provided is expired, revoked, malformed, "
297                "or invalid for other reasons"
298            ),
299            401,
300        ),
301        "insufficient_scope": (
302            "The request requires higher privileges than provided by the access token",
303            403,
304        ),
305    }
306
307    def __init__(self, code):
308        super().__init__()
309        self.code = code
310        error_tuple = self.errors.get(code, ("", ""))
311        self.description = error_tuple[0]
312        self.status = error_tuple[1]