feat(api): add scoped persistent access token infrastructure

alembic_users/versions/6d8f3cb50b71_add_persistent_access_tokens_table.py

@@ -0,0 +1,64 @@
+"""add persistent access tokens table
+
+Revision ID: 6d8f3cb50b71
+Revises: c1d2e3f4a5b6
+Create Date: 2026-03-31 00:25:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.engine.reflection import Inspector
+
+from gramps_webapi.auth.sql_guid import GUID
+
+
+# revision identifiers, used by Alembic.
+revision = "6d8f3cb50b71"
+down_revision = "c1d2e3f4a5b6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)
+    tables = inspector.get_table_names()
+    if "access_tokens" in tables:
+        # If table already exists, do nothing
+        return None
+
+    op.create_table(
+        "access_tokens",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column("user_id", GUID(), nullable=False),
+        sa.Column("scope", sa.String(length=64), nullable=False),
+        sa.Column("token_hash", sa.String(length=64), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column("revoked_at", sa.DateTime(), nullable=True),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
+        sa.UniqueConstraint("user_id", "scope", name="uq_access_tokens_user_scope"),
+    )
+    op.create_index("ix_access_tokens_user_id", "access_tokens", ["user_id"], unique=False)
+    op.create_index("ix_access_tokens_scope", "access_tokens", ["scope"], unique=False)
+    op.create_index(
+        "ix_access_tokens_token_hash", "access_tokens", ["token_hash"], unique=True
+    )
+
+
+def downgrade():
+    op.drop_index("ix_access_tokens_token_hash", table_name="access_tokens")
+    op.drop_index("ix_access_tokens_scope", table_name="access_tokens")
+    op.drop_index("ix_access_tokens_user_id", table_name="access_tokens")
+    op.drop_table("access_tokens")

gramps_webapi/api/__init__.py

@@ -34,6 +34,7 @@
     BookmarkResource,
     BookmarksResource,
 )
+from .resources.access_tokens import UserAccessTokenResource
 from .resources.chat import ChatResource
 from .resources.citations import CitationResource, CitationsResource
 from .resources.config import ConfigResource, ConfigsResource
@@ -551,6 +552,12 @@ def register_endpt(
 # User
 register_endpt(UsersResource, "/users/", "users", tags=["Users"])
 register_endpt(UserResource, "/users/<string:user_name>/", "user", tags=["Users"])
+register_endpt(
+    UserAccessTokenResource,
+    "/users/-/access-tokens/<string:scope>/",
+    "user_access_token",
+    tags=["Users"],
+)
 register_endpt(
     UserRegisterResource,
     "/users/<string:user_name>/register/",

gramps_webapi/api/resources/access_tokens.py

@@ -0,0 +1,97 @@
+#
+# Gramps Web API - A RESTful API for the Gramps genealogy program
+#
+# Copyright (C) 2026      Gramps Web contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+
+"""Persistent access token resources."""
+
+from flask_jwt_extended import get_jwt_identity
+from marshmallow import Schema
+from webargs import fields
+
+from ...auth import (
+    get_name,
+    has_user_access_token,
+    normalize_access_token_scope,
+    revoke_user_access_token,
+    rotate_user_access_token,
+)
+from ...auth.const import PERM_EDIT_OWN_USER
+from ..auth import require_permissions
+from ..blueprint import api_blueprint
+from ..util import abort_with_message
+from . import ProtectedResource
+
+
+class AccessTokenStatusSchema(Schema):
+    """Response schema for persistent access token status."""
+
+    active = fields.Boolean(
+        required=True,
+        metadata={"description": "Whether a token is currently active."},
+    )
+
+
+class AccessTokenCreateSchema(Schema):
+    """Response schema for newly created or rotated persistent token."""
+
+    active = fields.Boolean(
+        required=True,
+        metadata={"description": "Whether a token is currently active."},
+    )
+    token = fields.Str(
+        required=True,
+        metadata={"description": "Newly created persistent token value."},
+    )
+
+
+class UserAccessTokenResource(ProtectedResource):
+    """Resource for managing current user's persistent tokens by scope."""
+
+    def _get_user_name(self) -> str:
+        user_id = get_jwt_identity()
+        try:
+            return get_name(user_id)
+        except ValueError:
+            abort_with_message(401, "User not found for token ID")
+            raise  # unreachable
+
+    def _validate_scope(self, scope: str) -> str:
+        try:
+            return normalize_access_token_scope(scope)
+        except ValueError as exc:
+            abort_with_message(422, str(exc))
+            raise  # unreachable
+
+    @api_blueprint.response(200, AccessTokenStatusSchema())
+    def get(self, scope: str):
+        """Get persistent token status for current user and scope."""
+        require_permissions([PERM_EDIT_OWN_USER])
+        scope = self._validate_scope(scope)
+        user_name = self._get_user_name()
+        active = has_user_access_token(user_name, scope)
+        return {"active": active}, 200
+
+    @api_blueprint.response(200, AccessTokenCreateSchema())
+    def post(self, scope: str):
+        """Create or rotate persistent token for current user and scope."""
+        require_permissions([PERM_EDIT_OWN_USER])
+        scope = self._validate_scope(scope)
+        user_name = self._get_user_name()
+        token = rotate_user_access_token(user_name, scope)
+        return {"active": True, "token": token}, 200
+
+    @api_blueprint.response(200, AccessTokenStatusSchema())
+    def delete(self, scope: str):
+        """Revoke persistent token for current user and scope."""
+        require_permissions([PERM_EDIT_OWN_USER])
+        scope = self._validate_scope(scope)
+        user_name = self._get_user_name()
+        revoke_user_access_token(user_name, scope)
+        return {"active": False}, 200

gramps_webapi/auth/__init__.py

@@ -22,6 +22,7 @@
 
 import secrets
 import uuid
+from hashlib import sha256
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Sequence, Set, Union
 
@@ -34,7 +35,13 @@
 
 
 from ..const import DB_CONFIG_ALLOWED_KEYS
-from .const import PERMISSIONS, PERM_USE_CHAT, ROLE_ADMIN, ROLE_OWNER
+from .const import (
+    ACCESS_TOKEN_SCOPES,
+    PERMISSIONS,
+    PERM_USE_CHAT,
+    ROLE_ADMIN,
+    ROLE_OWNER,
+)
 from .passwords import hash_password, verify_password
 from .sql_guid import GUID
 
@@ -254,6 +261,100 @@ def get_user_details(username: str) -> Optional[Dict[str, Any]]:
     return _get_user_detail(user)
 
 
+def normalize_access_token_scope(scope: str) -> str:
+    """Validate and normalize a persistent access token scope."""
+    normalized_scope = (scope or "").strip().casefold()
+    if normalized_scope not in ACCESS_TOKEN_SCOPES:
+        raise ValueError("Unsupported access token scope")
+    return normalized_scope
+
+
+def _hash_access_token(token: str) -> str:
+    """Return deterministic SHA-256 hash for a persistent access token."""
+    return sha256(token.encode("utf-8")).hexdigest()
+
+
+def has_user_access_token(username: str, scope: str) -> bool:
+    """Return whether an active persistent access token exists for user+scope."""
+    scope = normalize_access_token_scope(scope)
+    query = user_db.session.query(User)  # pylint: disable=no-member
+    user = query.filter_by(name=username).scalar()
+    if user is None:
+        raise ValueError("User does not exist")
+    query = user_db.session.query(AccessToken)  # pylint: disable=no-member
+    access_token = query.filter(  # pylint: disable=no-member
+        AccessToken.user_id == user.id,
+        AccessToken.scope == scope,
+        AccessToken.token_hash.isnot(None),
+        AccessToken.revoked_at.is_(None),
+    ).scalar()
+    return access_token is not None
+
+
+def rotate_user_access_token(username: str, scope: str) -> str:
+    """Generate and persist a new token value for user+scope."""
+    scope = normalize_access_token_scope(scope)
+    query = user_db.session.query(User)  # pylint: disable=no-member
+    user = query.filter_by(name=username).scalar()
+    if user is None:
+        raise ValueError("User does not exist")
+    query = user_db.session.query(AccessToken)  # pylint: disable=no-member
+    access_token = query.filter_by(user_id=user.id, scope=scope).scalar()
+    for _ in range(5):
+        token = secrets.token_urlsafe(32)
+        token_hash = _hash_access_token(token)
+        if access_token is None:
+            access_token = AccessToken(user_id=user.id, scope=scope)
+            user_db.session.add(access_token)  # pylint: disable=no-member
+        access_token.token_hash = token_hash
+        access_token.revoked_at = None
+        access_token.updated_at = datetime.utcnow()
+        try:
+            user_db.session.commit()  # pylint: disable=no-member
+            return token
+        except IntegrityError:
+            user_db.session.rollback()  # pylint: disable=no-member
+            # Retry if token collided, or if concurrent insert happened.
+            access_token = query.filter_by(user_id=user.id, scope=scope).scalar()
+    raise ValueError("Could not generate a unique access token")
+
+
+def revoke_user_access_token(username: str, scope: str) -> None:
+    """Revoke token value for user+scope."""
+    scope = normalize_access_token_scope(scope)
+    query = user_db.session.query(User)  # pylint: disable=no-member
+    user = query.filter_by(name=username).scalar()
+    if user is None:
+        raise ValueError("User does not exist")
+    query = user_db.session.query(AccessToken)  # pylint: disable=no-member
+    access_token = query.filter_by(user_id=user.id, scope=scope).scalar()
+    if access_token is None:
+        return
+    access_token.token_hash = None
+    access_token.revoked_at = datetime.utcnow()
+    access_token.updated_at = datetime.utcnow()
+    user_db.session.commit()  # pylint: disable=no-member
+
+
+def get_user_from_access_token(token: str, scope: str) -> Optional["User"]:
+    """Return user matching persistent access token value and scope."""
+    if not token:
+        return None
+    scope = normalize_access_token_scope(scope)
+    token_hash = _hash_access_token(token)
+    query = user_db.session.query(User)  # pylint: disable=no-member
+    return (
+        query.join(AccessToken, AccessToken.user_id == User.id)
+        .filter(
+            AccessToken.scope == scope,
+            AccessToken.token_hash == token_hash,
+            AccessToken.token_hash.isnot(None),
+            AccessToken.revoked_at.is_(None),
+        )
+        .scalar()
+    )
+
+
 def get_all_user_details(
     tree: Optional[str],
     include_treeless=False,
@@ -545,6 +646,37 @@ def __repr__(self):
         return f"<User(name='{self.name}', fullname='{self.fullname}')>"
 
 
+class AccessToken(user_db.Model):  # type: ignore
+    """Persistent user access token table class for sqlalchemy."""
+
+    __tablename__ = "access_tokens"
+
+    id = mapped_column(sa.Integer, primary_key=True, autoincrement=True)
+    user_id = mapped_column(
+        GUID, sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    scope = mapped_column(sa.String(64), nullable=False, index=True)
+    token_hash = mapped_column(sa.String(64), nullable=True, unique=True, index=True)
+    created_at = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+    updated_at = mapped_column(
+        sa.DateTime, nullable=False, server_default=sa.func.now()
+    )
+    revoked_at = mapped_column(sa.DateTime, nullable=True)
+
+    __table_args__ = (
+        sa.UniqueConstraint("user_id", "scope", name="uq_access_tokens_user_scope"),
+    )
+
+    def __repr__(self):
+        """Return string representation of instance."""
+        return (
+            f"<AccessToken(user_id='{self.user_id}', scope='{self.scope}', "
+            f"revoked_at='{self.revoked_at}')>"
+        )
+
+
 class Config(user_db.Model):  # type: ignore
     """Config table class for sqlalchemy."""
 

gramps_webapi/auth/const.py

@@ -135,3 +135,9 @@
 SCOPE_CONF_EMAIL = "confirm_email"
 SCOPE_CREATE_ADMIN = "create_admin"
 SCOPE_CREATE_OWNER = "create_owner"
+
+# persistent access token scopes
+ACCESS_TOKEN_SCOPE_ANNIVERSARIES_ICS = "anniversaries_ics"
+ACCESS_TOKEN_SCOPES = {
+    ACCESS_TOKEN_SCOPE_ANNIVERSARIES_ICS,
+}