diff options
Diffstat (limited to 'gn3/auth/authorisation/oauth2')
| -rw-r--r-- | gn3/auth/authorisation/oauth2/__init__.py | 1 | ||||
| -rw-r--r-- | gn3/auth/authorisation/oauth2/oauth2client.py | 234 | ||||
| -rw-r--r-- | gn3/auth/authorisation/oauth2/oauth2token.py | 133 | ||||
| -rw-r--r-- | gn3/auth/authorisation/oauth2/resource_server.py | 19 |
4 files changed, 0 insertions, 387 deletions
diff --git a/gn3/auth/authorisation/oauth2/__init__.py b/gn3/auth/authorisation/oauth2/__init__.py deleted file mode 100644 index d083773..0000000 --- a/gn3/auth/authorisation/oauth2/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""OAuth2 modules.""" diff --git a/gn3/auth/authorisation/oauth2/oauth2client.py b/gn3/auth/authorisation/oauth2/oauth2client.py deleted file mode 100644 index dc54a41..0000000 --- a/gn3/auth/authorisation/oauth2/oauth2client.py +++ /dev/null @@ -1,234 +0,0 @@ -"""OAuth2 Client model.""" -import json -import datetime -from uuid import UUID -from typing import Sequence, Optional, NamedTuple - -from pymonad.maybe import Just, Maybe, Nothing - -from gn3.auth import db - -from gn3.auth.authorisation.errors import NotFoundError -from gn3.auth.authorisation.users import User, users, user_by_id, same_password - -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 - 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 same_password(client_secret, self.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. - - Acceptable client types: - * public: Unable to use registered client secrets, e.g. browsers, apps - on mobile devices. - * confidential: able to securely authenticate with authorisation server - e.g. being able to keep their registered client secret safe. - """ - 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_type", []) - - 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, - user: Optional[User] = None) -> 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() - the_user = user - if result: - if not bool(the_user): - try: - the_user = user_by_id(conn, result["user_id"]) - except NotFoundError as _nfe: - the_user = None - - return Just( - OAuth2Client(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"]), - the_user))# type: ignore[arg-type] - - return Nothing - -def client_by_id_and_secret(conn: db.DbConnection, client_id: UUID, - client_secret: str) -> OAuth2Client: - """Retrieve a client by its ID and secret""" - with db.cursor(conn) as cursor: - cursor.execute( - "SELECT * FROM oauth2_clients WHERE client_id=?", - (str(client_id),)) - row = cursor.fetchone() - if bool(row) and same_password(client_secret, row["client_secret"]): - return OAuth2Client( - client_id, client_secret, - datetime.datetime.fromtimestamp(row["client_id_issued_at"]), - datetime.datetime.fromtimestamp( - row["client_secret_expires_at"]), - json.loads(row["client_metadata"]), - user_by_id(conn, UUID(row["user_id"]))) - - raise NotFoundError("Could not find client with the given credentials.") - -def save_client(conn: db.DbConnection, the_client: OAuth2Client) -> OAuth2Client: - """Persist the client details into the database.""" - with db.cursor(conn) as cursor: - query = ( - "INSERT INTO oauth2_clients " - "(client_id, client_secret, client_id_issued_at, " - "client_secret_expires_at, client_metadata, user_id) " - "VALUES " - "(:client_id, :client_secret, :client_id_issued_at, " - ":client_secret_expires_at, :client_metadata, :user_id) " - "ON CONFLICT (client_id) DO UPDATE SET " - "client_secret=:client_secret, " - "client_id_issued_at=:client_id_issued_at, " - "client_secret_expires_at=:client_secret_expires_at, " - "client_metadata=:client_metadata, user_id=:user_id") - cursor.execute( - query, - { - "client_id": str(the_client.client_id), - "client_secret": the_client.client_secret, - "client_id_issued_at": ( - the_client.client_id_issued_at.timestamp()), - "client_secret_expires_at": ( - the_client.client_secret_expires_at.timestamp()), - "client_metadata": json.dumps(the_client.client_metadata), - "user_id": str(the_client.user.user_id) - }) - return the_client - -def oauth2_clients(conn: db.DbConnection) -> tuple[OAuth2Client, ...]: - """Fetch a list of all OAuth2 clients.""" - with db.cursor(conn) as cursor: - cursor.execute("SELECT * FROM oauth2_clients") - clients_rs = cursor.fetchall() - the_users = { - usr.user_id: usr for usr in users( - conn, tuple({UUID(result["user_id"]) for result in clients_rs})) - } - return tuple(OAuth2Client(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"]), - the_users[UUID(result["user_id"])]) - for result in clients_rs) - -def delete_client(conn: db.DbConnection, the_client: OAuth2Client) -> OAuth2Client: - """Delete the given client from the database""" - with db.cursor(conn) as cursor: - params = (str(the_client.client_id),) - cursor.execute("DELETE FROM authorisation_code WHERE client_id=?", - params) - cursor.execute("DELETE FROM oauth2_tokens WHERE client_id=?", params) - cursor.execute("DELETE FROM oauth2_clients WHERE client_id=?", params) - return the_client diff --git a/gn3/auth/authorisation/oauth2/oauth2token.py b/gn3/auth/authorisation/oauth2/oauth2token.py deleted file mode 100644 index bb19039..0000000 --- a/gn3/auth/authorisation/oauth2/oauth2token.py +++ /dev/null @@ -1,133 +0,0 @@ -"""OAuth2 Token""" -import uuid -import datetime -from typing import NamedTuple, Optional - -from pymonad.maybe import Just, Maybe, Nothing - -from gn3.auth import db - -from gn3.auth.authorisation.errors import NotFoundError -from gn3.auth.authorisation.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: 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 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: - def __identity__(val): - return val - try: - the_user = user_by_id(conn, uuid.UUID(rset["user_id"])) - except NotFoundError as _nfe: - the_user = None - the_client = client(conn, uuid.UUID(rset["client_id"]), the_user) - - if the_client.is_just() and bool(the_user): - 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"], - revoked=(rset["revoked"] == 1), - issued_at=datetime.datetime.fromtimestamp( - rset["issued_at"]), - expires_in=rset["expires_in"], - user=the_user))# type: ignore[arg-type] - - 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 (:token_id, :client_id, " - ":token_type, :access_token, :refresh_token, :scope, :revoked, " - ":issued_at, :expires_in, :user_id) " - "ON CONFLICT (token_id) DO UPDATE SET " - "refresh_token=:refresh_token, revoked=:revoked, " - "expires_in=:expires_in " - "WHERE token_id=:token_id"), - { - "token_id": str(token.token_id), - "client_id": str(token.client.client_id), - "token_type": token.token_type, - "access_token": token.access_token, - "refresh_token": token.refresh_token, - "scope": token.scope, - "revoked": 1 if token.revoked else 0, - "issued_at": int(token.issued_at.timestamp()), - "expires_in": token.expires_in, - "user_id": str(token.user.user_id) - }) diff --git a/gn3/auth/authorisation/oauth2/resource_server.py b/gn3/auth/authorisation/oauth2/resource_server.py deleted file mode 100644 index e806dc5..0000000 --- a/gn3/auth/authorisation/oauth2/resource_server.py +++ /dev/null @@ -1,19 +0,0 @@ -"""Protect the resources endpoints""" - -from flask import current_app as app -from authlib.oauth2.rfc6750 import BearerTokenValidator as _BearerTokenValidator -from authlib.integrations.flask_oauth2 import ResourceProtector - -from gn3.auth import db -from gn3.auth.authorisation.oauth2.oauth2token import token_by_access_token - -class BearerTokenValidator(_BearerTokenValidator): - """Extends `authlib.oauth2.rfc6750.BearerTokenValidator`""" - def authenticate_token(self, token_string: str): - with db.connection(app.config["AUTH_DB"]) as conn: - return token_by_access_token(conn, token_string).maybe(# type: ignore[misc] - None, lambda tok: tok) - -require_oauth = ResourceProtector() - -require_oauth.register_token_validator(BearerTokenValidator()) |