fischer-agentkit/src/agentkit/server/auth/providers/base.py

110 lines
4.3 KiB
Python

"""AuthProvider protocol — pluggable authentication backend contract.
The :class:`AuthProvider` Protocol defines the minimal surface area the
auth subsystem needs from any authentication backend (Local today, OIDC
/ SAML / LDAP tomorrow). Routes and admin endpoints call only these
methods and never touch the underlying user store directly. Adding a new
IdP integration is a matter of writing a new adapter that satisfies
this Protocol.
Design notes:
- **No sync surface for password verification**: the only entry point is
:meth:`authenticate` which takes the plaintext password. Internally each
adapter chooses how to verify (bcrypt locally, redirect to IdP, etc.).
- **No persistence responsibility**: providers return :class:`User` objects
but do not manage sessions, refresh tokens, or session state. That is
the route + SessionService's job (see :mod:`agentkit.server.auth.session`).
- **Audit-friendly**: :attr:`name` is written to ``auth_sessions.auth_provider``
on every login, so the source of every session is traceable. This is
what enables "list sessions by provider" admin queries and future
cross-IdP policy enforcement.
- **runtime_checkable**: the Protocol is decorated so that
``isinstance(provider, AuthProvider)`` works at runtime, enabling
defensive checks in tests and in DI wiring.
"""
from __future__ import annotations
from typing import Protocol, runtime_checkable
from .user import User
@runtime_checkable
class AuthProvider(Protocol):
"""All authentication backends must implement this surface.
The route layer only calls the methods below. It is intentionally
unaware of whether the backing store is SQLite, an OIDC IdP, LDAP,
or anything else.
"""
name: str
"""Identifier for this provider, written to ``auth_sessions.auth_provider``.
Convention: ``local`` for the local SQLite + bcrypt backend,
``oidc-<idp-name>`` for OIDC (e.g. ``oidc-keycloak``,
``oidc-feishu``), ``saml`` / ``ldap`` for the other planned
integrations. This value is the stable contract for audit
traceability — do not rename without a migration plan.
"""
async def authenticate(self, *, username: str, password: str) -> User:
"""Verify ``username`` + ``password`` and return the :class:`User`.
Args:
username: The submitted username (or, for OIDC, the IdP
subject id — but that's a future adapter's concern).
password: The submitted plaintext password.
Returns:
The matched :class:`User` on success.
Raises:
InvalidCredentials: if the user does not exist, the password
is wrong, or the user is inactive. Callers MUST NOT
distinguish between these three cases in the error
message returned to the client (timing-attack /
username-enumeration mitigation).
"""
...
async def get_user_by_id(self, user_id: str) -> User | None:
"""Look up a :class:`User` by primary key.
Used by:
- Admin endpoints that need to display user info by id
- Session validation in the cold-start / whoami path
- Audit log enrichment
Returns ``None`` if no user exists with this id (or the user
is inactive — convention: inactive users are "not found" from
the auth layer's perspective).
"""
...
async def sync_user_attributes(self, user_id: str) -> None:
"""Refresh user attributes (department / email / title) from the source of truth.
- :class:`LocalAuthProvider`: no-op (attributes are managed locally).
- OIDC adapter (future): pull the latest profile from the IdP and
write back to the local ``users`` table.
Implementations that have nothing to sync should still define
this method (returning ``None``) so the contract is uniform.
"""
...
async def revoke_user(self, user_id: str) -> None:
"""Disable a user account (e.g. on termination or lock-out).
- :class:`LocalAuthProvider`: ``UPDATE users SET is_active = 0``
- OIDC adapter (future): call the IdP's disable API
The admin endpoint that calls this does NOT also need to
revoke the user's active sessions — that is the
:class:`SessionService`'s job, called separately.
"""
...