feat(oauth): personal access token schema and model

Author: rboni-dk

testgen/api/oauth/models.py

@@ -21,10 +21,10 @@
 from testgen import settings
 from testgen.api.deps import db_session
 from testgen.api.oauth.login import render_login_page
-from testgen.api.oauth.models import OAuth2Client
 from testgen.api.oauth.server import TestGenAuthorizationServer
 from testgen.common.auth import create_jwt_token, decode_jwt_token, verify_password
 from testgen.common.models import get_current_session
+from testgen.common.models.oauth import OAuth2Client
 from testgen.common.models.user import User
 
 LOG = logging.getLogger("testgen")

testgen/api/oauth/routes.py

@@ -2,7 +2,6 @@
 
 Grant types:
 - Authorization Code + PKCE (for MCP clients)
-- Client Credentials (for automation scripts)
 - Refresh Token (for token renewal)
 
 All DB operations use get_current_session() for thread-local session access.
@@ -13,15 +12,14 @@
 from typing import ClassVar
 
 from authlib.oauth2.rfc6749 import AuthorizationServer, JsonRequest, OAuth2Request, grants
-from authlib.oauth2.rfc6749.errors import InvalidGrantError
 from authlib.oauth2.rfc7009 import RevocationEndpoint
 from authlib.oauth2.rfc7636 import CodeChallenge
 from sqlalchemy import select
 
 from testgen import settings
-from testgen.api.oauth.models import OAuth2AuthorizationCode, OAuth2Client, OAuth2Token
 from testgen.common.auth import create_jwt_token
 from testgen.common.models import get_current_session
+from testgen.common.models.oauth import OAuth2AuthorizationCode, OAuth2Client, OAuth2Token
 from testgen.common.models.user import User
 
 
@@ -85,24 +83,6 @@ def revoke_old_credential(self, credential):
         credential.access_token_revoked_at = int(time.time())
 
 
-class ClientCredentialsGrant(grants.ClientCredentialsGrant):
-    """Client credentials grant that resolves the client's owner as the token user.
-
-    Ensures every token has a real User identity — no "ghost" usernames.
-    """
-
-    def validate_token_request(self):
-        super().validate_token_request()
-        client = self.request.client
-        if not client.user_id:
-            raise InvalidGrantError(description="Client has no registered owner.")
-        session = get_current_session()
-        owner = session.scalars(select(User).where(User.id == client.user_id)).first()
-        if owner is None:
-            raise InvalidGrantError(description="Client owner no longer exists.")
-        self.request.user = owner
-
-
 class TestGenRevocationEndpoint(RevocationEndpoint):
     def query_token(self, token_string, token_type_hint):
         session = get_current_session()
@@ -183,7 +163,6 @@ def create_authorization_server() -> TestGenAuthorizationServer:
     """Create and configure the authorization server with all grant types."""
     server = TestGenAuthorizationServer()
     server.register_grant(AuthorizationCodeGrant, [CodeChallenge(required=True)])
-    server.register_grant(ClientCredentialsGrant)
     server.register_grant(RefreshTokenGrant)
     server.register_endpoint(TestGenRevocationEndpoint)
     server.register_token_generator("default", _generate_bearer_token)

testgen/api/oauth/server.py

@@ -4,8 +4,11 @@
 
 import bcrypt
 import jwt
+from sqlalchemy import func, select
 
 from testgen import settings
+from testgen.common.models.oauth import OAuth2Token
+from testgen.common.models.user import User
 
 LOG = logging.getLogger("testgen")
 
@@ -45,11 +48,6 @@ def authorize_token(token_str: str, username: str, session):
 
     Shared implementation for API and MCP authorization.
     """
-    from sqlalchemy import func, select
-
-    from testgen.api.oauth.models import OAuth2Token
-    from testgen.common.models.user import User
-
     user = session.scalars(select(User).where(func.lower(User.username) == func.lower(username))).first()
     if user is None:
         raise AuthError("User not found")

testgen/common/auth.py

@@ -0,0 +1,90 @@
+import time
+from enum import StrEnum
+
+from authlib.integrations.sqla_oauth2 import (
+    OAuth2AuthorizationCodeMixin,
+    OAuth2ClientMixin,
+    OAuth2TokenMixin,
+)
+from sqlalchemy import Column, ForeignKey, String, case, func
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.ext.hybrid import hybrid_property
+
+from testgen import settings
+from testgen.common.models import Base
+
+
+class OAuth2ClientType(StrEnum):
+    """How an OAuth2 client was provisioned.
+
+    PAT clients are created by the authenticated personal-access-token path and owned by
+    a user; EXTERNAL clients are dynamically registered (MCP apps, automation scripts).
+    """
+
+    PAT = "pat"
+    EXTERNAL = "external"
+
+
+class PersonalAccessTokenStatus(StrEnum):
+    """Display status of a personal access token. Revoked takes precedence over expired."""
+
+    ACTIVE = "active"
+    REVOKED = "revoked"
+    EXPIRED = "expired"
+
+
+class OAuth2Client(Base, OAuth2ClientMixin):
+    __tablename__ = "oauth2_clients"
+
+    id = Column(postgresql.UUID(as_uuid=True), primary_key=True, server_default="gen_random_uuid()")
+    client_type = Column(String(20), nullable=False, default=OAuth2ClientType.EXTERNAL)
+
+    # Override to widen — JWTs can exceed 255 chars
+    # (the mixin defines client_id as VARCHAR(48) which is fine)
+
+
+class OAuth2AuthorizationCode(Base, OAuth2AuthorizationCodeMixin):
+    __tablename__ = "oauth2_authorization_codes"
+
+    id = Column(postgresql.UUID(as_uuid=True), primary_key=True, server_default="gen_random_uuid()")
+    user_id = Column(postgresql.UUID(as_uuid=True), ForeignKey("auth_users.id", ondelete="CASCADE"), nullable=False)
+
+
+class OAuth2Token(Base, OAuth2TokenMixin):
+    __tablename__ = "oauth2_tokens"
+
+    id = Column(postgresql.UUID(as_uuid=True), primary_key=True, server_default="gen_random_uuid()")
+    user_id = Column(postgresql.UUID(as_uuid=True), ForeignKey("auth_users.id", ondelete="CASCADE"), nullable=True)
+
+    # Override to allow longer JWTs as access tokens
+    access_token = Column(String(2048), unique=True, nullable=False)
+
+    # Set only for personal access tokens; NULL for tokens from the auth-code / MCP flows
+    name = Column(String(255), nullable=True)
+
+    def is_refresh_token_active(self) -> bool:
+        if self.refresh_token_revoked_at:
+            return False
+        expires_at = self.issued_at + settings.REFRESH_TOKEN_EXPIRES_IN
+        return expires_at >= time.time()
+
+    @hybrid_property
+    def status(self) -> PersonalAccessTokenStatus:
+        """Personal access token status, derived from revocation and expiry.
+
+        Instance side reads the app clock; the SQL expression reads the DB clock —
+        both are wall-clock "now", evaluated independently.
+        """
+        if self.access_token_revoked_at:
+            return PersonalAccessTokenStatus.REVOKED
+        if self.issued_at + self.expires_in < time.time():
+            return PersonalAccessTokenStatus.EXPIRED
+        return PersonalAccessTokenStatus.ACTIVE
+
+    @status.expression  # type: ignore[no-redef]
+    def status(cls):
+        return case(
+            (cls.access_token_revoked_at > 0, PersonalAccessTokenStatus.REVOKED.value),
+            (cls.issued_at + cls.expires_in < func.extract("epoch", func.now()), PersonalAccessTokenStatus.EXPIRED.value),
+            else_=PersonalAccessTokenStatus.ACTIVE.value,
+        )

testgen/common/models/oauth.py

@@ -533,6 +533,17 @@ def _ssl_files_present() -> bool:
 Lifetime of OAuth access and refresh tokens.
 """
 
+PAT_DEFAULT_LIFETIME_SECONDS: int = 31_536_000  # 365 days
+"""
+Default maximum lifetime a personal access token can be created with.
+"""
+
+PAT_MAX_LIFETIME_SECONDS: int = 63_072_000  # 2 years (730 days)
+"""
+Absolute ceiling on the admin-configurable personal access token maximum lifetime.
+The configured maximum cannot exceed this.
+"""
+
 JWT_HASHING_KEY_B64: str = getenv("TG_JWT_HASHING_KEY")
 """
 Random key used to sign/verify the authentication token

testgen/settings.py

@@ -742,12 +742,12 @@ CREATE INDEX ix_pm_role ON project_memberships(role);
 
 CREATE TABLE oauth2_clients (
     id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-    user_id         UUID REFERENCES auth_users(id) ON DELETE SET NULL,
     client_id       VARCHAR(48) NOT NULL UNIQUE,
     client_secret   VARCHAR(120),
     client_id_issued_at     INTEGER NOT NULL DEFAULT 0,
     client_secret_expires_at INTEGER NOT NULL DEFAULT 0,
-    client_metadata TEXT NOT NULL DEFAULT '{}'
+    client_metadata TEXT NOT NULL DEFAULT '{}',
+    client_type     VARCHAR(20) NOT NULL DEFAULT 'external'
 );
 CREATE INDEX idx_oauth2_clients_client_id ON oauth2_clients(client_id);
 
@@ -778,7 +778,8 @@ CREATE TABLE oauth2_tokens (
     issued_at       INTEGER NOT NULL DEFAULT EXTRACT(EPOCH FROM NOW())::INTEGER,
     access_token_revoked_at  INTEGER NOT NULL DEFAULT 0,
     refresh_token_revoked_at INTEGER NOT NULL DEFAULT 0,
-    expires_in      INTEGER NOT NULL DEFAULT 0
+    expires_in      INTEGER NOT NULL DEFAULT 0,
+    name            VARCHAR(255)
 );
 CREATE INDEX idx_oauth2_tokens_refresh_token ON oauth2_tokens(refresh_token);
 

testgen/template/dbsetup/030_initialize_new_schema_structure.sql

@@ -0,0 +1,15 @@
+SET SEARCH_PATH TO {SCHEMA_NAME};
+
+-- Personal access tokens are stored as oauth2_tokens rows; the name labels each
+-- one in the user's token list. NULL marks a non-PAT token (auth-code / MCP flows).
+ALTER TABLE oauth2_tokens ADD COLUMN name VARCHAR(255);
+
+-- Classify clients so personal-access-token clients are distinguishable from
+-- dynamically-registered external clients (MCP apps, automation). Existing rows
+-- all predate PATs and are externally registered, so they default to 'external'.
+ALTER TABLE oauth2_clients ADD COLUMN client_type VARCHAR(20) NOT NULL DEFAULT 'external';
+
+-- Drop the client owner: it was only ever read by the client-credentials grant
+-- (removed). User identity now rides the token (oauth2_tokens.user_id + the JWT
+-- username), so a client needs no owner. The FK constraint drops with the column.
+ALTER TABLE oauth2_clients DROP COLUMN user_id;

testgen/template/dbupgrade/0198_incremental_upgrade.sql

@@ -84,6 +84,7 @@ def render(log_level: int = logging.INFO):
                         support_email=settings.SUPPORT_EMAIL,
                         global_context=is_global_context,
                         is_global_admin=session.auth.user_has_permission("global_admin") and bool(application.global_admin_paths),
+                        account_path=application.account_path,
                     )
 
         application.router.run()

testgen/ui/app.py

@@ -49,13 +49,14 @@
 
 
 class Application(singleton.Singleton):
-    def __init__(self, auth_class: Authentication, logo: plugins.Logo, router: Router, menu: Menu, logger: logging.Logger, global_admin_paths: frozenset[str]) -> None:
+    def __init__(self, auth_class: Authentication, logo: plugins.Logo, router: Router, menu: Menu, logger: logging.Logger, global_admin_paths: frozenset[str], account_path: str | None) -> None:
         self.auth_class = auth_class
         self.logo = logo
         self.router = router
         self.menu = menu
         self.logger = logger
         self.global_admin_paths = global_admin_paths
+        self.account_path = account_path
 
 
 def run(log_level: int = logging.INFO) -> Application:
@@ -91,4 +92,5 @@ def run(log_level: int = logging.INFO) -> Application:
         ),
         logger=LOG,
         global_admin_paths=frozenset(page.path for page in pages if page.permission == "global_admin"),
+        account_path=next((page.path for page in pages if page.is_account_page), None),
     )