agent-platform/services/auth-service/app/application/services.py

from dataclasses import dataclass
from datetime import datetime

from app.db.models import ApiKey, Role, RoleAssignment, RolePermissionBinding, User
from app.domain.repositories import (
    ApiKeyRepository,
    RoleAssignmentRepository,
    RolePermissionBindingRepository,
    RoleRepository,
    UserRepository,
)
from app.infrastructure.api_keys import generate_api_key, get_api_key_prefix, hash_api_key
from app.infrastructure.passwords import verify_password
from app.infrastructure.tokens import TokenError, issue_access_token, verify_access_token


@dataclass(frozen=True)
class LoginResult:
    access_token: str
    expires_time: datetime
    user: User


@dataclass(frozen=True)
class TokenVerificationResult:
    active: bool
    user_id: str | None = None
    username: str | None = None
    expires_time: datetime | None = None
    reason: str | None = None


@dataclass(frozen=True)
class PermissionCheckResult:
    allowed: bool
    reason: str
    matched_role_ids: list[str]


class AuthApplicationService:
    def __init__(
        self,
        *,
        user_repository: UserRepository,
        role_repository: RoleRepository,
        assignment_repository: RoleAssignmentRepository,
        permission_binding_repository: RolePermissionBindingRepository,
        api_key_repository: ApiKeyRepository,
        token_secret: str) -> None:
        self.user_repository = user_repository
        self.role_repository = role_repository
        self.assignment_repository = assignment_repository
        self.permission_binding_repository = permission_binding_repository
        self.api_key_repository = api_key_repository
        self.token_secret = token_secret

    def login(self, *, username: str, password: str) -> LoginResult | None:
        user = self.user_repository.get_by_username(username=username)
        if user is None or user.status != "active":
            return None
        if not verify_password(password, user.password_hash):
            return None

        self.user_repository.touch_last_login_time(user_id=user.id)
        access_token, expires_time = issue_access_token(
            user_id=user.id,
            secret=self.token_secret)
        return LoginResult(
            access_token=access_token,
            expires_time=expires_time,
            user=user)

    def verify_token(self, *, access_token: str) -> TokenVerificationResult:
        try:
            token_payload = verify_access_token(access_token, secret=self.token_secret)
        except TokenError as exc:
            return TokenVerificationResult(active=False, reason=str(exc))

        expires_time_raw = token_payload["expires_time"]
        user = self.user_repository.get_by_id(user_id=token_payload["user_id"])
        if user is None or user.status != "active":
            return TokenVerificationResult(active=False, reason="user_not_active")

        return TokenVerificationResult(
            active=True,
            user_id=user.id,
            username=user.username,
            expires_time=datetime.fromisoformat(expires_time_raw.removesuffix("Z")))

    def list_users(self) -> list[User]:
        return self.user_repository.list_all()

    def list_users_page(
        self,
        *,
        page: int,
        page_size: int,
        keyword: str | None) -> tuple[list[User], int]:
        return self.user_repository.list_page(
            offset=(page - 1) * page_size,
            limit=page_size,
            keyword=keyword)

    def list_roles(self) -> list[Role]:
        return self.role_repository.list_all()

    def list_roles_page(
        self,
        *,
        page: int,
        page_size: int,
        keyword: str | None) -> tuple[list[Role], int]:
        return self.role_repository.list_page(
            offset=(page - 1) * page_size,
            limit=page_size,
            keyword=keyword)

    def list_assignments(self, *, user_id: str) -> list[RoleAssignment]:
        return self.assignment_repository.list_by_user(user_id=user_id)

    def list_role_permission_bindings(
        self,
        *,
        role_id: str,
        page: int,
        page_size: int) -> tuple[list[RolePermissionBinding], int]:
        return self.permission_binding_repository.list_by_role(
            role_id=role_id,
            offset=(page - 1) * page_size,
            limit=page_size)

    def add_role_permission_binding(
        self,
        *,
        role_id: str,
        permission: str,
        scope_type: str | None,
        scope_id: str | None) -> RolePermissionBinding:
        return self.permission_binding_repository.create(
            role_id=role_id,
            permission=permission,
            scope_type=scope_type,
            scope_id=scope_id)

    def remove_role_permission_binding(self, *, binding_id: str) -> bool:
        return self.permission_binding_repository.delete(binding_id=binding_id)

    def list_api_keys_page(
        self,
        *,
        page: int,
        page_size: int,
        keyword: str | None) -> tuple[list[ApiKey], int]:
        return self.api_key_repository.list_page(
            offset=(page - 1) * page_size,
            limit=page_size,
            keyword=keyword)

    def create_api_key(
        self,
        *,
        name: str,
        scopes: str | None,
        expires_time: datetime | None) -> tuple[ApiKey, str]:
        secret = generate_api_key()
        entity = self.api_key_repository.create(
            name=name,
            key_prefix=get_api_key_prefix(secret),
            key_hash=hash_api_key(secret),
            scopes=scopes,
            expires_time=expires_time)
        return entity, secret

    def revoke_api_key(self, *, api_key_id: str) -> ApiKey | None:
        return self.api_key_repository.revoke(api_key_id=api_key_id)

    def check_permission(
        self,
        *,
        user_id: str,
        permission: str,
        scope_type: str | None,
        scope_id: str | None) -> PermissionCheckResult:
        user = self.user_repository.get_by_id(user_id=user_id)
        if user is None or user.status != "active":
            return PermissionCheckResult(
                allowed=False,
                reason="user_not_active",
                matched_role_ids=[])

        assignments = self.assignment_repository.list_by_user(user_id=user_id)
        matched_role_ids: list[str] = []
        now = datetime.utcnow()
        for assignment in assignments:
            if assignment.status != "active":
                continue
            if assignment.expires_time is not None and assignment.expires_time <= now:
                continue
            if not self._scope_matches(
                assignment=assignment,
                scope_type=scope_type,
                scope_id=scope_id):
                continue
            role = self.role_repository.get_by_id(role_id=assignment.role_id)
            if role is None or role.status != "active":
                continue
            if self._role_has_permission(
                role,
                permission,
                scope_type=scope_type,
                scope_id=scope_id):
                matched_role_ids.append(role.id)

        return PermissionCheckResult(
            allowed=bool(matched_role_ids),
            reason="matched" if matched_role_ids else "permission_not_found",
            matched_role_ids=matched_role_ids)

    def _permission_matches(self, permissions: list[str], requested_permission: str) -> bool:
        if "*" in permissions or requested_permission in permissions:
            return True
        return any(
            permission.endswith(":*")
            and requested_permission.startswith(permission.removesuffix("*"))
            for permission in permissions
        )

    def _scope_matches(
        self,
        *,
        assignment: RoleAssignment,
        scope_type: str | None,
        scope_id: str | None) -> bool:
        if assignment.scope_type is None and assignment.scope_id is None:
            return True
        return assignment.scope_type == scope_type and assignment.scope_id == scope_id

    def _role_has_permission(
        self,
        role: Role,
        requested_permission: str,
        *,
        scope_type: str | None,
        scope_id: str | None) -> bool:
        bindings = self.permission_binding_repository.list_all_by_role(role_id=role.id)
        return any(
            (binding.scope_type is None and binding.scope_id is None
             or (binding.scope_type == scope_type and binding.scope_id == scope_id))
            and
            self._permission_matches([binding.permission], requested_permission)
            for binding in bindings
        )