From 498c79f434856aa68e9a883247ea69a22256fce6 Mon Sep 17 00:00:00 2001 From: Tom van der Lee Date: Wed, 22 Jul 2026 14:51:44 +0200 Subject: Added auth layer --- authentication/mixins.py | 54 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) create mode 100644 authentication/mixins.py (limited to 'authentication/mixins.py') diff --git a/authentication/mixins.py b/authentication/mixins.py new file mode 100644 index 0000000..3d6a91f --- /dev/null +++ b/authentication/mixins.py @@ -0,0 +1,54 @@ +import json +import time + +from authlib.integrations.sqla_oauth2 import OAuth2ClientMixin as BaseOAuth2ClientMixin, OAuth2AuthorizationCodeMixin as BaseOAuth2AuthorizationCodeMixin, OAuth2TokenMixin as BaseOAuth2TokenMixin +from sqlalchemy.sql.sqltypes import Text +from sqlmodel import Field + + +class OAuth2ClientMixin(BaseOAuth2ClientMixin): + client_id: str = Field(max_length=48, index=True, nullable=True) + client_secret: str = Field(max_length=120, nullable=True) + client_id_issued_at: int = Field(nullable=False, default=0) + client_secret_expires_at: int = Field(nullable=False, default=0) + client_metadata_field: str = Field(sa_type=Text, sa_column_kwargs={ + 'name': 'client_metadata' + }, nullable=True) + + _client_metadata = None + + @property + def client_metadata(self) -> dict: + if self.client_metadata_field: + return json.loads(self.client_metadata_field) + return {} + + def set_client_metadata(self, metadata: dict): + self.client_metadata_field = json.dumps(metadata) + + +class OAuth2AuthorizationCodeMixin(BaseOAuth2AuthorizationCodeMixin): + code: str = Field(max_length=120, unique=True, nullable=False) + client_id: str = Field(max_length=48, nullable=True) + redirect_uri: str = Field(sa_type=Text, default="", nullable=True) + response_type: str = Field(sa_type=Text, default="", nullable=True) + scope: str = Field(sa_type=Text, default="", nullable=True) + nonce: str = Field(sa_type=Text, nullable=True) + auth_time: int = Field(nullable=False, default_factory=lambda: int(time.time())) + acr: str = Field(sa_type=Text, nullable=True) + amr: str = Field(sa_type=Text, nullable=True) + + code_challenge: str = Field(sa_type=Text, nullable=True) + code_challenge_method: str = Field(max_length=48, nullable=True) + + +class OAuth2TokenMixin(BaseOAuth2TokenMixin): + client_id: str = Field(max_length=48, nullable=True) + token_type: str = Field(max_length=40, nullable=True) + access_token: str = Field(max_length=255, unique=True, nullable=False) + refresh_token: str = Field(max_length=255, index=True, nullable=True) + scope: str = Field(sa_type=Text, default="", nullable=True) + issued_at: int = Field(nullable=False, default_factory=lambda: int(time.time())) + access_token_revoked_at: int = Field(nullable=False, default=0) + refresh_token_revoked_at: int = Field(nullable=False, default=0) + expires_in: int = Field(nullable=False, default=0) -- cgit v1.2.3