task_credentials

`task_credentials` ¶

Operator-facing credential management for live tasks.

Two verbs ride alongside the rest of terok task ...:

revoke_credentials — nuke every phantom token bound to a (project, task) pair. The in-flight request that already reached the broker still completes; every subsequent call from the container's agent gets a 401. Operator response to "audit shows this task is misbehaving" without killing the container itself, so the forensic state stays intact for inspection.
audit_credentials — filter the broker's append-only credential-audit JSONL by the task's (scope, subject) pair, with optional provider / time / tail filters.

Both translate the operator's project/task identifiers into the opaque (scope, subject) pair the sandbox actually keys on. scope is the project name; subject is the task id. The mapping lives here rather than in terok-sandbox because the sandbox makes no claim about what those labels identify.

`all = ['audit_credentials', 'revoke_credentials']` `module-attribute` ¶

`revoke_credentials(project_name, task_id)` ¶

Revoke every phantom token bound to (project_name, task_id).

The DB-side delete takes effect on the next request the agent issues — the broker has no in-memory token cache, so an in-flight proxy call already past lookup_token completes on its real credential. That's the intended semantics: block further use, don't kill in-flight TCP.

Parameters:

Name	Type	Description	Default
`project_name`	`str`	Project id (becomes the token's `scope`).	required
`task_id`	`str`	Task id (becomes the token's `subject`).	required

Returns:

Type	Description
`int`	Number of phantom tokens removed (zero if there were none —
`int`	idempotent).

Source code in src/terok/lib/domain/task_credentials.py

def revoke_credentials(project_name: str, task_id: str) -> int:
    """Revoke every phantom token bound to ``(project_name, task_id)``.

    The DB-side delete takes effect on the next request the agent
    issues — the broker has no in-memory token cache, so an in-flight
    proxy call already past
    [`lookup_token`][terok_sandbox.vault.store.db.CredentialDB.lookup_token]
    completes on its real credential.  That's the intended semantics:
    block further use, don't kill in-flight TCP.

    Args:
        project_name: Project id (becomes the token's ``scope``).
        task_id: Task id (becomes the token's ``subject``).

    Returns:
        Number of phantom tokens removed (zero if there were none —
        idempotent).
    """
    from .vault import vault_db

    with vault_db() as db:
        return db.revoke_tokens(project_name, task_id)

`audit_credentials(project_name, task_id, *, provider=None, since=None, tail=None)` ¶

Yield audit lines for (project_name, task_id) after filtering.

Reads the broker's credential_audit_log_path JSONL, drops lines whose scope / subject don't match the target task, applies the optional provider / time filters, and keeps only the last tail survivors when set.

Parameters:

Name	Type	Description	Default
`project_name`	`str`	Project id (matched against the line's `scope`).	required
`task_id`	`str`	Task id (matched against the line's `subject`).	required
`provider`	`str \| None`	When set, drop lines whose `provider` field differs.	`None`
`since`	`datetime \| None`	When set, drop lines whose `ts` parses earlier than this datetime. Lines with malformed timestamps are kept (the sanitiser invariant guarantees printable ASCII, but mismatched ISO-8601 shapes are still possible from future schema changes).	`None`
`tail`	`int \| None`	When set, yield only the last `N` matching lines.	`None`

Yields:

Type	Description
`dict`	Decoded JSON dicts, one per surviving line.

Source code in src/terok/lib/domain/task_credentials.py

def audit_credentials(
    project_name: str,
    task_id: str,
    *,
    provider: str | None = None,
    since: datetime | None = None,
    tail: int | None = None,
) -> Iterator[dict]:
    """Yield audit lines for ``(project_name, task_id)`` after filtering.

    Reads the broker's
    [`credential_audit_log_path`][terok_sandbox.SandboxConfig.credential_audit_log_path]
    JSONL, drops lines whose ``scope`` / ``subject`` don't match the
    target task, applies the optional provider / time filters, and
    keeps only the last ``tail`` survivors when set.

    Args:
        project_name: Project id (matched against the line's ``scope``).
        task_id: Task id (matched against the line's ``subject``).
        provider: When set, drop lines whose ``provider`` field differs.
        since: When set, drop lines whose ``ts`` parses earlier than
            this datetime.  Lines with malformed timestamps are kept
            (the sanitiser invariant guarantees printable ASCII, but
            mismatched ISO-8601 shapes are still possible from
            future schema changes).
        tail: When set, yield only the last ``N`` matching lines.

    Yields:
        Decoded JSON dicts, one per surviving line.
    """
    from ..core.config import make_sandbox_config

    audit_path = make_sandbox_config().credential_audit_log_path
    matches: Iterable[dict] = _filtered_lines(
        audit_path, project_name, task_id, provider=provider, since=since
    )
    if tail is not None and tail > 0:
        # Materialise to slice from the tail; full-history scans rarely
        # outpace the operator's terminal so the cost is negligible.
        matches = list(matches)[-tail:]
    yield from matches

task_credentials