diff options
Diffstat (limited to 'gn3/auth/authentication/oauth2/models')
| -rw-r--r-- | gn3/auth/authentication/oauth2/models/__init__.py | 0 | ||||
| -rw-r--r-- | gn3/auth/authentication/oauth2/models/oauth2client.py | 141 | ||||
| -rw-r--r-- | gn3/auth/authentication/oauth2/models/oauth2token.py | 119 |
3 files changed, 260 insertions, 0 deletions
diff --git a/gn3/auth/authentication/oauth2/models/__init__.py b/gn3/auth/authentication/oauth2/models/__init__.py new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/gn3/auth/authentication/oauth2/models/__init__.py diff --git a/gn3/auth/authentication/oauth2/models/oauth2client.py b/gn3/auth/authentication/oauth2/models/oauth2client.py new file mode 100644 index 0000000..2ee7858 --- /dev/null +++ b/gn3/auth/authentication/oauth2/models/oauth2client.py @@ -0,0 +1,141 @@ +"""OAuth2 Client model.""" +import json +import uuid +import datetime +from typing import NamedTuple, Sequence + +from pymonad.maybe import Just, Maybe, Nothing + +from gn3.auth import db +from gn3.auth.authentication.users import User, user_by_id + +class OAuth2Client(NamedTuple): + """ + Client to the OAuth2 Server. + + This is defined according to the mixin at + https://docs.authlib.org/en/latest/specs/rfc6749.html#authlib.oauth2.rfc6749.ClientMixin + """ + client_id: uuid.UUID + client_secret: str + client_id_issued_at: datetime.datetime + client_secret_expires_at: datetime.datetime + client_metadata: dict + user: User + + def check_client_secret(self, client_secret: str) -> bool: + """Check whether the `client_secret` matches this client.""" + return self.client_secret == client_secret + + @property + def token_endpoint_auth_method(self) -> str: + """Return the token endpoint authorisation method.""" + return self.client_metadata.get("token_endpoint_auth_method", ["none"]) + + @property + def client_type(self) -> str: + """Return the token endpoint authorisation method.""" + return self.client_metadata.get("client_type", "public") + + def check_endpoint_auth_method(self, method: str, endpoint: str) -> bool: + """ + Check if the client supports the given method for the given endpoint. + + Acceptable methods: + * none: Client is a public client and does not have a client secret + * client_secret_post: Client uses the HTTP POST parameters + * client_secret_basic: Client uses HTTP Basic + """ + if endpoint == "token": + return (method in self.token_endpoint_auth_method + and method == "client_secret_post") + if endpoint in ("introspection", "revoke"): + return (method in self.token_endpoint_auth_method + and method == "client_secret_basic") + return False + + @property + def id(self):# pylint: disable=[invalid-name] + """Return the client_id.""" + return self.client_id + + @property + def grant_types(self) -> Sequence[str]: + """ + Return the grant types that this client supports. + + Valid grant types: + * authorisation_code + * implicit + * client_credentials + * password + """ + return self.client_metadata.get("grant_types", []) + + def check_grant_type(self, grant_type: str) -> bool: + """ + Validate that client can handle the given grant types + """ + return grant_type in self.grant_types + + @property + def redirect_uris(self) -> Sequence[str]: + """Return the redirect_uris that this client supports.""" + return self.client_metadata.get('redirect_uris', []) + + def check_redirect_uri(self, redirect_uri: str) -> bool: + """ + Check whether the given `redirect_uri` is one of the expected ones. + """ + return redirect_uri in self.redirect_uris + + @property + def response_types(self) -> Sequence[str]: + """Return the response_types that this client supports.""" + return self.client_metadata.get("response_types", []) + + def check_response_type(self, response_type: str) -> bool: + """Check whether this client supports `response_type`.""" + return response_type in self.response_types + + @property + def scope(self) -> Sequence[str]: + """Return valid scopes for this client.""" + return tuple(set(self.client_metadata.get("scope", []))) + + def get_allowed_scope(self, scope: str) -> str: + """Return list of scopes in `scope` that are supported by this client.""" + if not bool(scope): + return "" + requested = scope.split() + return " ".join(sorted(set( + scp for scp in requested if scp in self.scope))) + + def get_client_id(self): + """Return this client's identifier.""" + return self.client_id + + def get_default_redirect_uri(self) -> str: + """Return the default redirect uri""" + return self.client_metadata.get("default_redirect_uri", "") + +def client(conn: db.DbConnection, client_id: uuid.UUID) -> Maybe: + """Retrieve a client by its ID""" + with db.cursor(conn) as cursor: + cursor.execute( + "SELECT * FROM oauth2_clients WHERE client_id=?", (str(client_id),)) + result = cursor.fetchone() + if result: + return Just( + OAuth2Client(uuid.UUID(result["client_id"]), + result["client_secret"], + datetime.datetime.fromtimestamp( + result["client_id_issued_at"]), + datetime.datetime.fromtimestamp( + result["client_secret_expires_at"]), + json.loads(result["client_metadata"]), + user_by_id( # type: ignore[misc] + conn, uuid.UUID(result["user_id"])).maybe( + None, lambda usr: usr))) + + return Nothing diff --git a/gn3/auth/authentication/oauth2/models/oauth2token.py b/gn3/auth/authentication/oauth2/models/oauth2token.py new file mode 100644 index 0000000..70421b4 --- /dev/null +++ b/gn3/auth/authentication/oauth2/models/oauth2token.py @@ -0,0 +1,119 @@ +"""OAuth2 Token""" +import uuid +import datetime +from typing import NamedTuple, Optional, Sequence + +from pymonad.maybe import Just, Maybe, Nothing + +from gn3.auth import db +from gn3.auth.authentication.users import User, user_by_id + +from .oauth2client import client, OAuth2Client + +class OAuth2Token(NamedTuple): + """Implement Tokens for OAuth2.""" + token_id: uuid.UUID + client: OAuth2Client + token_type: str + access_token: str + refresh_token: Optional[str] + scope: Sequence[str] + revoked: bool + issued_at: datetime.datetime + expires_in: int + user: User + + @property + def expires_at(self) -> datetime.datetime: + """Return the time when the token expires.""" + return self.issued_at + datetime.timedelta(seconds=self.expires_in) + + def check_client(self, client: OAuth2Client) -> bool:# pylint: disable=[redefined-outer-name] + """Check whether the token is issued to given `client`.""" + return client.client_id == self.client.client_id + + def get_expires_in(self) -> int: + """Return the `expires_in` value for the token.""" + return self.expires_in + + def get_scope(self) -> str: + """Return the valid scope for the token.""" + return " ".join(self.scope) + + def is_expired(self) -> bool: + """Check whether the token is expired.""" + return self.expires_at < datetime.datetime.now() + + def is_revoked(self): + """Check whether the token has been revoked.""" + return self.revoked + +def __token_from_resultset__(conn: db.DbConnection, rset) -> Maybe: + the_client = client(conn, uuid.UUID(rset["client_id"])) + the_user = user_by_id(conn, uuid.UUID(rset["user_id"])) + __identity__ = lambda val: val + + if the_client.is_just() and the_user.is_just(): + return Just(OAuth2Token(token_id=uuid.UUID(rset["token_id"]), + client=the_client.maybe(None, __identity__), + token_type=rset["token_type"], + access_token=rset["access_token"], + refresh_token=rset["refresh_token"], + scope=rset["scope"].split(None), + revoked=(rset["revoked"] == 1), + issued_at=datetime.datetime.fromtimestamp( + rset["issued_at"]), + expires_in=rset["expires_in"], + user=the_user.maybe(None, __identity__))) + + return Nothing + +def token_by_access_token(conn: db.DbConnection, token_str: str) -> Maybe: + """Retrieve token by its token string""" + with db.cursor(conn) as cursor: + cursor.execute("SELECT * FROM oauth2_tokens WHERE access_token=?", + (token_str,)) + res = cursor.fetchone() + if res: + return __token_from_resultset__(conn, res) + + return Nothing + +def token_by_refresh_token(conn: db.DbConnection, token_str: str) -> Maybe: + """Retrieve token by its token string""" + with db.cursor(conn) as cursor: + cursor.execute( + "SELECT * FROM oauth2_tokens WHERE refresh_token=?", + (token_str,)) + res = cursor.fetchone() + if res: + return __token_from_resultset__(conn, res) + + return Nothing + +def revoke_token(token: OAuth2Token) -> OAuth2Token: + """ + Return a new token derived from `token` with the `revoked` field set to + `True`. + """ + return OAuth2Token( + token_id=token.token_id, client=token.client, + token_type=token.token_type, access_token=token.access_token, + refresh_token=token.refresh_token, scope=token.scope, revoked=True, + issued_at=token.issued_at, expires_in=token.expires_in, user=token.user) + +def save_token(conn: db.DbConnection, token: OAuth2Token) -> None: + """Save/Update the token.""" + with db.cursor(conn) as cursor: + cursor.execute( + "INSERT INTO oauth2_tokens VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (str(token.token_id), str(token.client.client_id), token.token_type, + token.access_token, token.refresh_token, token.scope, + 1 if token.revoked else 0, int(token.issued_at.timestamp()), + token.expires_in, str(token.user.user_id))) + ## If already exists + # cursor.execute( + # ("UPDATE oauth2_tokens SET refresh_token=?, revoked=?, " + # "expires_in=? WHERE token_id=?"), + # (token.refresh_token, token.scope, 1 if token.revoked else 0, + # token.expires_in, str(token.token_id))) |