Permissions¶

async def get_current_user(
    db: graph.Pool,
    credentials: security.HTTPAuthorizationCredentials
    | None = fastapi.Depends(oauth2_scheme),  # noqa: B008
) -> AuthContext:
    """FastAPI dependency to get the current authenticated user.

    Supports both JWT and API key authentication. API keys are
    detected by the 'ik_' prefix.

    Args:
        db: Graph database connection (injected by FastAPI).
        credentials: HTTP Bearer credentials from Authorization
            header.

    Returns:
        AuthContext with user and permissions

    Raises:
        fastapi.HTTPException: If authentication fails

    """
    if not credentials:
        raise fastapi.HTTPException(
            status_code=401,
            detail='Missing authentication credentials',
            headers={'WWW-Authenticate': 'Bearer'},
        )
    return await _authenticate_token(
        db, credentials.credentials, auth_settings()
    )

async def get_current_user_cookie_fallback(
    db: graph.Pool,
    request: fastapi.Request,
    credentials: security.HTTPAuthorizationCredentials
    | None = fastapi.Depends(oauth2_scheme),  # noqa: B008
) -> AuthContext:
    """Like ``get_current_user`` but also accepts the access cookie.

    Browser subresource requests (e.g. ``<img src>``) cannot set an
    ``Authorization`` header, so for the upload-serving GET endpoints the
    access token is also read from the :data:`ACCESS_COOKIE_NAME` cookie
    when the header is absent. The Bearer header still takes precedence.

    Raises:
        fastapi.HTTPException: If neither a header nor cookie token is
            present.

    """
    token = (
        credentials.credentials
        if credentials
        else request.cookies.get(ACCESS_COOKIE_NAME)
    )
    if not credentials and token and core.is_agent_api_token(token):
        # A runner sends it as a header only.
        token = None
    if not token:
        raise fastapi.HTTPException(
            status_code=401,
            detail='Missing authentication credentials',
            headers={'WWW-Authenticate': 'Bearer'},
        )
    return await _authenticate_token(db, token, auth_settings())

def require_permission(
    permission: str,
    *,
    allow_cookie: bool = False,
) -> typing.Callable[[AuthContext], collections.abc.Awaitable[AuthContext]]:
    """
    Create a FastAPI dependency that enforces a specific permission.

    The returned dependency validates the current request's AuthContext:
    admin users bypass the check; otherwise the dependency ensures the
    required permission is present and returns the AuthContext when
    allowed.

    Parameters:
        permission (str): Permission name to require (e.g.,
            "blueprint:read").
        allow_cookie (bool): When True, the access token may also be
            supplied via the :data:`ACCESS_COOKIE_NAME` cookie instead of
            the Authorization header. Intended only for endpoints loaded
            as browser subresources (``<img src>``), which cannot send a
            Bearer header.

    Returns:
        Callable[[AuthContext], Awaitable[AuthContext]]: A dependency
            callable that returns the current AuthContext when the user
            has the required permission.

    Raises:
        fastapi.HTTPException: Raised with status code 403 if the
            current user lacks the required permission.
    """
    resolver = (
        get_current_user_cookie_fallback if allow_cookie else get_current_user
    )

    async def check_permission(
        auth: typing.Annotated[AuthContext, fastapi.Depends(resolver)],
    ) -> AuthContext:
        """Enforce that the principal has the required permission.

        Admin users bypass checks. Service accounts never bypass.

        Returns:
            AuthContext when the permission is granted.

        """
        if auth.is_admin:
            return auth

        if permission not in auth.permissions:
            LOGGER.warning(
                'Permission denied: principal=%s permission=%s',
                auth.principal_name,
                permission,
            )
            raise fastapi.HTTPException(
                status_code=403,
                detail=f'Permission denied: {permission} required',
            )
        return auth

    # Expose the required permission for schema generation. Permission
    # enforcement lives in a closure, so it is otherwise invisible to
    # ``app.openapi()``; ``openapi.py`` reads this attribute to stamp
    # ``x-imbi-permission`` for AI consumers building per-caller
    # toolsets.
    check_permission.imbi_permission = permission  # type: ignore[attr-defined]

    return check_permission

async def authenticate_jwt(
    db: graph.Graph,
    token: str,
    auth_settings: settings.Auth,
) -> AuthContext:
    """
    Validate a JWT, load the corresponding user and their permissions,
    and return an AuthContext.

    Parameters:
        db: Graph database connection.
        token (str): JWT access token string.
        auth_settings (settings.Auth): Configuration used to decode
            and validate the token.

    Returns:
        AuthContext: Authentication context containing the resolved
            user, the token's `jti` as `session_id`, `auth_method`
            set to `'jwt'`, and the user's permission set.

    Raises:
        fastapi.HTTPException: On token expiry, invalid token, invalid
            token type, revoked token, missing subject, user not found,
            or inactive user account.
    """
    try:
        # Decode and validate token
        claims = core.verify_token(token, auth_settings)
    except jwt.ExpiredSignatureError as err:
        raise fastapi.HTTPException(
            status_code=401, detail='Token has expired'
        ) from err
    except jwt.InvalidTokenError as err:
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid token'
        ) from err

    # Check token type
    if claims.get('type') != 'access':
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid token type'
        )

    # Check if token is revoked
    jti = claims.get('jti')
    query = 'MATCH (t:TokenMetadata {{jti: {jti}}}) RETURN t.revoked'
    records = await db.execute(query, {'jti': jti}, columns=['revoked'])
    if records:
        revoked = graph.parse_agtype(records[0].get('revoked'))
        if revoked:
            raise fastapi.HTTPException(
                status_code=401, detail='Token revoked'
            )

    # Load principal (user or service account)
    subject = claims.get('sub')
    if not subject:
        raise fastapi.HTTPException(
            status_code=401, detail='Token missing subject'
        )

    auth_method = claims.get('auth_method', 'jwt')

    if auth_method == 'client_credentials':
        # Service account token
        sa_results = await db.match(models.ServiceAccount, {'slug': subject})
        if not sa_results:
            raise fastapi.HTTPException(
                status_code=401,
                detail='Service account not found',
            )
        sa = sa_results[0]

        if not sa.is_active:
            raise fastapi.HTTPException(
                status_code=401,
                detail='Service account is inactive',
            )

        perms = await load_principal_permissions(
            db, 'ServiceAccount', 'slug', subject
        )
        # The same rule as an API key: a token without scopes has every
        # permission of the account's roles, and a token with scopes has
        # only the ones that are in both.
        scopes = str(claims.get('scope') or '').split()
        if scopes:
            perms &= set(scopes)
        return AuthContext(
            service_account=sa,
            session_id=jti,
            auth_method='client_credentials',
            permissions=perms,
        )

    # Standard user token
    user_results = await db.match(models.User, {'email': subject})
    if not user_results:
        raise fastapi.HTTPException(status_code=401, detail='User not found')
    user = user_results[0]

    # Check if user is active
    if not user.is_active:
        raise fastapi.HTTPException(
            status_code=401, detail='User account is inactive'
        )

    # Load permissions
    perms = await load_principal_permissions(db, 'User', 'email', subject)
    identities = await _load_user_identities(db, user.id)

    return AuthContext(
        user=user,
        session_id=jti,
        auth_method='jwt',
        channel='slack' if claims.get('channel') == 'slack' else None,
        permissions=perms,
        identities=identities,
    )

async def authenticate_api_key(
    db: graph.Graph,
    key: str,
    auth_settings: settings.Auth,
) -> AuthContext:
    """
    Validate an API key, load the corresponding user and their
    permissions, and return an AuthContext (Phase 5).

    API keys have the format: ik_<key_id>_<secret>

    Parameters:
        db: Graph database connection.
        key (str): Full API key string.
        auth_settings (settings.Auth): Configuration for validation.

    Returns:
        AuthContext: Authentication context with user, key_id as
            session_id, auth_method set to 'api_key', and filtered
            permissions based on key scopes.

    Raises:
        fastapi.HTTPException: On invalid format, revoked key, expired
            key, invalid secret, or inactive user.
    """
    # Parse API key format: ik_<id>_<secret>
    parts = key.split('_', 2)
    if len(parts) != 3 or parts[0] != 'ik':
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid API key format'
        )

    key_id = f'ik_{parts[1]}'
    key_secret = parts[2]

    # Fetch API key and owner (User or ServiceAccount)
    query = (
        'MATCH (k:APIKey {{key_id: {key_id}}}) '
        'OPTIONAL MATCH (k)-[:OWNED_BY]->(u:User) '
        'OPTIONAL MATCH (k)-[:OWNED_BY]->(s:ServiceAccount) '
        'RETURN k, u, s'
    )
    records = await db.execute(
        query, {'key_id': key_id}, columns=['k', 'u', 's']
    )

    if not records:
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid or revoked API key'
        )

    api_key_data = graph.parse_agtype(records[0]['k'])
    user_data = graph.parse_agtype(records[0]['u'])
    sa_data = graph.parse_agtype(records[0]['s'])

    if not user_data and not sa_data:
        raise fastapi.HTTPException(
            status_code=401, detail='API key owner not found'
        )

    # Check if key is revoked
    if api_key_data.get('revoked', False):
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid or revoked API key'
        )

    # Check if key is expired -- AGE stores datetime as ISO strings
    expires_at = api_key_data.get('expires_at')
    if expires_at:
        if isinstance(expires_at, str):
            expires_at = datetime.datetime.fromisoformat(
                expires_at,
            )
        # A stored value with no offset parses naive, and comparing that to an
        # aware `now` raises `TypeError` -- which on this path would answer an
        # expired key with a 500 instead of a 401. Assume UTC, which is what
        # every writer of this property stores.
        if expires_at.tzinfo is None:
            expires_at = expires_at.replace(tzinfo=datetime.UTC)
        if expires_at < datetime.datetime.now(datetime.UTC):
            raise fastapi.HTTPException(
                status_code=401,
                detail='API key expired',
            )

    key_hash = api_key_data.get('key_hash')

    # M15: fast path -- a hot key (e.g. a CI bot looping requests)
    # would otherwise re-run Argon2 and the permission queries per
    # call. It comes after the revoked and expiry checks above, so
    # those take effect at once. ``_stamp_api_key_last_used`` runs
    # only on a miss, so ``last_used`` updates at most once per TTL.
    # The owner row is already loaded, so check it before a cache hit:
    # deactivating the owner then takes effect at once as well.
    owner = sa_data or user_data
    if not owner.get('is_active', True):
        raise fastapi.HTTPException(
            status_code=401,
            detail='Service account is inactive'
            if sa_data
            else 'User account is inactive',
        )

    cached = _api_key_cache_lookup(key, key_hash) if key_hash else None
    if cached is not None:
        return cached

    # Verify key secret (hashed). A row without `key_hash` is unverifiable,
    # which is a failed authentication and not a server fault -- subscripting
    # it would raise `KeyError` and answer a bad credential with a 500.
    if not key_hash or not await asyncio.to_thread(
        password.verify_password, key_secret, key_hash
    ):
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid or revoked API key'
        )

    # Resolve owner and permissions
    scopes = models.parse_scopes(
        api_key_data.get('scopes', []),
    )

    if sa_data:
        sa = models.ServiceAccount(**sa_data)
        if not sa.is_active:
            raise fastapi.HTTPException(
                status_code=401,
                detail='Service account is inactive',
            )
        all_perms = await load_principal_permissions(
            db, 'ServiceAccount', 'slug', sa.slug
        )
        filtered = all_perms.intersection(set(scopes)) if scopes else all_perms

        # Update last_used only after all validation passes
        await _stamp_api_key_last_used(db, key_id, auth_settings)

        ctx = AuthContext(
            service_account=sa,
            session_id=key_id,
            auth_method='api_key',
            permissions=filtered,
        )
        _api_key_cache_store(key, key_hash, ctx)
        return ctx

    user = models.User(**user_data)
    if not user.is_active:
        raise fastapi.HTTPException(
            status_code=401, detail='User account is inactive'
        )

    all_perms = await load_principal_permissions(
        db, 'User', 'email', user.email
    )
    filtered = all_perms.intersection(set(scopes)) if scopes else all_perms
    identities = await _load_user_identities(db, user.id)

    # Update last_used only after all validation passes
    await _stamp_api_key_last_used(db, key_id, auth_settings)

    ctx = AuthContext(
        user=user,
        session_id=key_id,
        auth_method='api_key',
        permissions=filtered,
        identities=identities,
    )
    _api_key_cache_store(key, key_hash, ctx)
    return ctx

async def authenticate_agent_api(
    db: graph.Graph, token: str, auth_settings: settings.Auth
) -> AuthContext:
    """Authenticate an agent API token as its service account, with the
    roles of that service account, in the token's organization only.

    Raises:
        fastapi.HTTPException: 401 when the token is not valid, has
            expired (its lease), or its service account is inactive.

    """
    try:
        claims = core.verify_agent_api_token(token, auth_settings)
    except jwt.ExpiredSignatureError as err:
        raise fastapi.HTTPException(
            status_code=401, detail='Token has expired'
        ) from err
    except jwt.InvalidTokenError as err:
        raise fastapi.HTTPException(
            status_code=401, detail='Invalid token'
        ) from err
    accounts = await db.match(models.ServiceAccount, {'slug': claims['sub']})
    if not accounts or not accounts[0].is_active:
        raise fastapi.HTTPException(
            status_code=401, detail='Service account is inactive'
        )
    return AuthContext(
        service_account=accounts[0],
        session_id=str(claims['jti']),
        auth_method='agent_api',
        organization_id=str(claims['org']),
        permissions=await load_principal_permissions(
            db, 'ServiceAccount', 'slug', str(claims['sub'])
        ),
    )

def accept_agent_api_tokens() -> None:
    """Accept agent API tokens (ADR 0022, harness plan P7) in this
    service. imbi-api calls it, and refuses them on its auth, token,
    service account, role, secret, and harness routes."""
    global _agent_api_accepted
    _agent_api_accepted = True

def clear_api_key_cache() -> None:
    """Drop every cached API-key AuthContext.

    Tests use this to keep cases independent.
    """
    _api_key_cache.clear()

async def load_principal_permissions(
    db: graph.Graph,
    label: PrincipalLabel,
    match_prop: PrincipalMatchProp,
    value: str,
    org_slug: str | None = None,
) -> set[str]:
    """Get permission names granted to a principal (user or SA).

    Collects permissions from the principal's organization memberships,
    following role inheritance via ``INHERITS_FROM`` and ``GRANTS``
    relationships. With ``org_slug``, only the role in that
    organization counts; without it, the roles in all of them.

    Parameters:
        db: Graph database connection.
        label: Node label of the principal (``'User'`` or
            ``'ServiceAccount'``).
        match_prop: Property name on the principal used to match
            (``'email'`` for users, ``'slug'`` for service accounts).
        value: Value of ``match_prop`` identifying the principal.
        org_slug: The organization to resolve the permissions in.

    Returns:
        Set of permission names (for example, ``'blueprint:read'``).

    Raises:
        ValueError: If ``(label, match_prop)`` is not an allowed
            principal selector. ``label`` and ``match_prop`` are
            interpolated into the Cypher template, so this guard
            prevents Cypher injection via future non-constant inputs.
    """
    if (label, match_prop) not in _ALLOWED_PRINCIPAL_SELECTORS:
        raise ValueError(
            f'Unsupported principal selector: ({label!r}, {match_prop!r})'
        )
    # Apache AGE does not honor the zero-hop case in variable-length
    # paths (``*0..``), so the start role itself would be excluded
    # from the traversal. Collect ancestors with ``*1..`` then union
    # the start role back in.
    query = (
        f'MATCH (p:{label} {{{{{match_prop}: {{value}}}}}})'
        '-[m:MEMBER_OF]->(o:Organization) '
        + ('WHERE o.slug = {org_slug} ' if org_slug is not None else '')
        + 'MATCH (r:Role {{slug: m.role}}) '
        'OPTIONAL MATCH (r)-[:INHERITS_FROM*1..]->(ancestor:Role) '
        'WITH r, collect(DISTINCT ancestor) AS ancestors '
        'UNWIND ancestors + [r] AS role '
        'WITH DISTINCT role '
        'OPTIONAL MATCH (role)-[:GRANTS]->(perm:Permission) '
        'RETURN collect(DISTINCT perm.name)'
    )
    params = {'value': value}
    if org_slug is not None:
        params['org_slug'] = org_slug
    records = await db.execute(query, params, columns=['permissions'])
    if not records:
        return set()
    raw: typing.Any = graph.parse_agtype(records[0].get('permissions'))
    if isinstance(raw, list):
        return {
            item
            for item in typing.cast(list[str | typing.Any], raw)
            if isinstance(item, str)
        }
    return set()

async def load_all_permission_names(db: graph.Graph) -> set[str]:
    """Return the set of seeded ``Permission.name`` values.

    Used to validate API-key / SA-key / client-credential scope lists
    at creation time — a scope that doesn't correspond to a seeded
    permission would silently grant nothing (the auth path
    intersects scopes with the principal's actual permissions), but
    persisting bogus scopes pollutes the audit trail and hides
    typos.
    """
    records = await db.execute(
        'MATCH (p:Permission) RETURN p.name AS name',
        {},
        columns=['name'],
    )
    names: set[str] = set()
    for row in records:
        raw = graph.parse_agtype(row.get('name'))
        if isinstance(raw, str) and raw:
            names.add(raw)
    return names

async def validate_scopes(
    db: graph.Graph, scopes: collections.abc.Iterable[str]
) -> None:
    """Raise ``HTTPException(400)`` if any scope isn't a seeded Permission.

    An empty / all-scopes list (``[]``) is treated as "no restriction"
    by callers and passes validation trivially.
    """
    requested = {s for s in scopes if s}
    if not requested:
        return
    known = await load_all_permission_names(db)
    unknown = sorted(requested - known)
    if unknown:
        raise fastapi.HTTPException(
            status_code=400,
            detail=(
                'Unknown scope(s): '
                + ', '.join(unknown)
                + '. Scopes must match seeded Permission names.'
            ),
        )

def set_auth_settings_provider(
    provider: collections.abc.Callable[[], settings.Auth],
) -> None:
    """Register how *this module* resolves Auth settings.

    A member with an ``Auth`` subclass must call this so token
    verification reads the same singleton the member's own code does.
    Two singletons would each auto-generate a JWT secret when
    ``IMBI_AUTH_JWT_SECRET`` is unset, and tokens minted against one
    would fail verification against the other.

    Scoped to the authentication path, and deliberately not a fix for the
    split in general: :mod:`imbi.common.access_log` and
    :mod:`imbi.common.auth.encryption` still call
    ``settings.get_auth_settings`` directly, so they read imbi-common's
    singleton even in a process that registered a provider here. Closing
    that properly means one singleton for the whole process -- a
    registerable concrete class on ``settings.get_auth_settings``, or an
    injected dependency -- which is a larger change than the auth
    extraction warranted.

    This is a process-wide mutable hook, so it is order-dependent by
    construction: the last caller wins, and in a single-process test run
    an import is enough to set it.
    """
    global _auth_settings_provider
    _auth_settings_provider = provider

def auth_settings() -> settings.Auth:
    """Return the Auth settings this process authenticates against."""
    return _auth_settings_provider()