Authentication API¶
The OAuth2 layer: token models, the store protocol and its backends, the
headless code-exchange/refresh functions, the provider that clients consume,
the login callback listener, and the browser-login choreography shared by the
CLI and the MCP server's auth_login tool.
anafpy.auth.models ¶
Token model.
ANAF access tokens are JWTs valid 90 days; refresh tokens 365 days. Refresh rotates the refresh token (a new access and refresh token come back), so both are persisted. Expiry timestamps are computed from the JWTs themselves (cached per instance), so the persisted store can never drift from the tokens it holds.
TokenSet ¶
Bases: BaseModel
An access/refresh token pair; expiries are derived from the JWT exp claims.
Instances are treated as immutable: refresh rotation builds a new TokenSet
via :meth:from_token_response, which is what makes the cached expiry properties
safe.
access_expires_at
cached
property
¶
access_expires_at: float
Access expiry (epoch s): JWT exp, else expires_in, else 90 days.
refresh_expires_at
cached
property
¶
refresh_expires_at: float
Refresh expiry (epoch s): JWT exp, else the documented 365 days.
from_token_response
classmethod
¶
from_token_response(data: dict[str, Any], *, obtained_at: float | None = None) -> TokenSet
Build from ANAF's JSON token response (access_token, refresh_token).
access_expired ¶
access_expired(*, leeway: float = 300.0) -> bool
True if the access token is expired (or within leeway seconds of it).
refresh_expired ¶
refresh_expired() -> bool
True if the refresh token has likely expired (re-auth required).
anafpy.auth.store ¶
Token persistence.
TokenStore is the abstraction the rest of the library depends on;
KeyringTokenStore is the default backend — tokens live in the OS credential store
(macOS Keychain, Windows Credential Manager, Linux Secret Service/KWallet).
FileTokenStore (a JSON file with owner-only permissions, suitable for mounting as
a Docker volume) is the opt-out for headless hosts without a credential store.
TokenStore ¶
Bases: Protocol
Where token sets are loaded from and saved to.
MemoryTokenStore ¶
MemoryTokenStore(tokens: TokenSet | None = None)
In-process token store (tests, ephemeral runtimes).
FileTokenStore ¶
FileTokenStore(path: str | PathLike[str])
Bases: JsonFileStore[TokenSet]
JSON-file token store, written atomically with 0o600 permissions.
The Docker/headless opt-out from the keyring default; custody mechanics are
:class:~anafpy._store.JsonFileStore's, shared with the SPV session store.
KeyringTokenStore ¶
KeyringTokenStore(service: str = 'anafpy', username: str = 'tokens', *, chunk_size: int | None = None)
Token store in the OS credential store, via the keyring library.
The default backend. Uses the platform's native secret store — macOS Keychain,
Windows Credential Manager, or a Linux Secret Service/KWallet daemon. The token
set is stored as JSON under (service, username); a set longer than
chunk_size characters is split across continuation entries (username#1,
#2, ...), which is what makes Windows work at all (its blob cap is smaller
than one ANAF JWT). chunk_size=None picks the platform default: split on
Windows, one entry everywhere else.
Construction fails with :class:~anafpy.exceptions.AnafConfigError when no
usable OS backend exists (e.g. headless Linux without a Secret Service
daemon) — use the file backend there.
load ¶
load() -> TokenSet | None
The stored token set, or None when the credential store has none.
Raises:
| Type | Description |
|---|---|
AnafConfigError
|
entries exist but cannot be read or parsed — stay in the AnafError hierarchy instead of leaking a keyring/pydantic error. |
clear ¶
clear() -> None
Delete the token entries (including continuation chunks); no-op when absent.
Raises:
| Type | Description |
|---|---|
AnafConfigError
|
the credential store cannot be read or written. |
anafpy.auth.oauth ¶
ANAF OAuth2 endpoints and operations.
Verified against the official OAuth procedure PDF and a live probe (2026-06-28):
client auth is HTTP Basic (client_id:client_secret); token_content_type is
jwt (query for /authorize, body for /token); refresh is headless (no
certificate). See docs/anaf-reference/oauth/authentication.md.
build_authorize_url ¶
build_authorize_url(client_id: str, redirect_uri: str, *, state: str | None = None) -> str
Build the browser authorization URL (the cert step happens here).
anafpy.auth.provider ¶
Token lifecycle: hand out a valid access token, refreshing transparently.
TokenProvider is the batteries-included implementation over a TokenStore.
AnafAuth plugs it into httpx2 so every request carries a Bearer token and a 401
triggers a single refresh-and-retry.
TokenProvider ¶
TokenProvider(*, client_id: str, client_secret: str, store: TokenStore, http: AsyncClient | None = None)
Provides valid access tokens, refreshing and persisting as needed.
The TokenStore is the single source of truth: the provider keeps no token
state of its own. Every operation reads the freshest persisted set under the
lock, so a login or a refresh-token rotation performed by another process
sharing the store (the CLI, a second server) is picked up on the next call.
tokens
property
¶
tokens: TokenSet | None
The persisted token set, if authenticated (read-only snapshot).
access_token
async
¶
access_token() -> str
Return a currently-valid access token, refreshing if expired.
force_refresh
async
¶
force_refresh(*, stale: str | None = None) -> str
Refresh unconditionally (used after a 401); return the new access token.
Pass the access token the failing request carried as stale: when it has
already been replaced (a concurrent request or another process refreshed
first), the current token is returned without burning another rotation.
AnafAuth ¶
AnafAuth(provider: TokenProvider)
Bases: Auth
httpx2 auth flow: attach Bearer, and refresh-once on a 401.
anafpy.auth.callback ¶
Authorization-code capture for the one-time auth login bootstrap.
ANAF redirects the browser (after the certificate step) to the registered callback URL
with ?code=.... The ANAF developer portal rejects http:// callback URLs
(HTTP 400 at registration — verified 2026-07-02), so the registered URL is https://
and there are two ways to capture the code:
- Listener (:class:
CallbackListener/capture_authorization_code): a tiny local server on the callback URL's host/port. Plain HTTP by default (put a TLS terminator in front), or pass anssl.SSLContextto serve TLS directly with a certificate you supply. The listener binds on construction so the browser can be opened only once the port is actually listening — a fast redirect (cached certificate/session) must never outrun the bind. - Paste mode (
parse_redirect_url): run no listener at all. The browser lands on a connection error, but the address bar still holds the full redirect URL; the user pastes it (or just the code) into the CLI. Works everywhere, needs no certificate.
Both capture paths take an expected_state: the CLI binds a random OAuth state
to each login attempt, and a redirect that does not echo it back is rejected (the
listener answers 400 and keeps waiting) — so a forged redirect cannot inject an
attacker's authorization code into the flow (login CSRF).
CallbackListener ¶
CallbackListener(redirect_uri: str, *, ssl_context: SSLContext | None = None, expected_state: str | None = None)
A bound-and-listening receiver for the OAuth redirect.
Binding happens in the constructor (raising :class:AnafConfigError when the
host/port cannot be bound), so callers can open the browser only after the
listener is provably up, then :meth:wait for the redirect. Use as a context
manager (or call :meth:close) to stop the server.
When expected_state is given, only a redirect echoing that OAuth state
completes the wait; anything else (a forged redirect trying to inject a code,
login CSRF) is answered 400 and the listener keeps waiting for the real one.
wait ¶
wait(timeout: float = 180.0) -> str | None
Block until the redirect arrives; the code, or None on timeout.
Raises AnafAuthError on an OAuth error redirect.
parse_redirect_url ¶
parse_redirect_url(pasted: str, *, expected_state: str | None = None) -> str
Extract the authorization code from a pasted redirect URL (paste mode).
Accepts the full redirect URL from the browser address bar, a bare
code=...&... query string, or the code value alone. When
expected_state is given, a pasted URL must echo it back (a bare code —
a deliberate manual extraction — is exempt).
Raises AnafAuthError on an OAuth error redirect, a state mismatch,
or unrecognizable input.
capture_authorization_code ¶
capture_authorization_code(redirect_uri: str, *, timeout: float = 180.0, ssl_context: SSLContext | None = None, expected_state: str | None = None) -> str
Block until ANAF redirects to redirect_uri with a code; return that code.
Bind-and-wait in one call, for callers that already opened the browser (prefer
:class:CallbackListener to bind before opening it). Plain HTTP unless
ssl_context is given, in which case it serves TLS with that context (use
this when the browser must reach the callback over https:// and no
external TLS terminator is in front). expected_state is enforced as in
:class:CallbackListener.
Raises AnafAuthError on an OAuth error redirect or on timeout.
anafpy.auth.browser ¶
The browser-login choreography shared by the CLI and the MCP server.
Both interactive front-ends — anafpy auth login and the MCP server's
confirm-gated auth_login tool — run the same security-critical sequence:
a fresh OAuth state per attempt, the callback listener bound before the
browser opens (with a cached certificate/session the redirect can arrive within
a second, and ANAF's codes expire in ~60s), the code exchange, and the save to
the token store. :func:browser_login is that sequence's single home; nothing
past its two gates raises for a flow outcome — every way an attempt can end is
a member of the :data:BrowserLoginOutcome union, and each front-end translates
those into its own interaction model (interactive prints and paste fallbacks;
a structured tool result). :func:finish_login is the exchange-and-save tail,
also used on its own by the CLI's paste paths, which capture the code without a
listener.
BrowserLoginOutcome ¶
BrowserLoginOutcome = LoginCompleted | ListenerUnavailable | BrowserNotOpened | AuthorizationDenied | CallbackTimedOut | ExchangeFailed
Every way one browser-login attempt can end. Closed — match with
assert_never.
LoginCompleted ¶
Bases: BaseModel
The code was exchanged and the tokens saved to the store.
ListenerUnavailable ¶
Bases: BaseModel
The callback port could not be bound; the browser was not opened.
authorize_url and state let a front-end continue the same attempt
without a listener (open the browser itself, then capture by paste).
BrowserNotOpened ¶
Bases: BaseModel
No browser could be opened and require_browser was set.
The listener is already closed; authorize_url/state describe the
attempt for reporting or a paste continuation.
AuthorizationDenied ¶
Bases: BaseModel
ANAF redirected with an OAuth error — usually a cancelled certificate step.
CallbackTimedOut ¶
Bases: BaseModel
No redirect arrived within timeout.
The browser's address bar still holds the redirect after a late
authorization, so a paste continuation with the same state remains
possible.
ExchangeFailed ¶
Bases: BaseModel
The token endpoint refused the code or was unreachable (codes expire ~60s).
finish_login
async
¶
finish_login(*, client_id: str, client_secret: str, code: str, redirect_uri: str, store: TokenStore) -> TokenSet
Exchange a captured authorization code and persist the tokens.
The tail of every login: shared by :func:browser_login and the CLI's
paste paths, so the exchange parameters and the save cannot drift apart.
Raises:
| Type | Description |
|---|---|
AnafAuthError
|
if the token endpoint refuses the code. |
AnafTransportError
|
if the token endpoint is unreachable. |
browser_login
async
¶
browser_login(client_id: str, client_secret: str, redirect_uri: str, store: TokenStore, *, timeout: float = 180.0, ssl_context: SSLContext | None = None, require_browser: bool = False, open_browser: Callable[[str], bool] | None = None) -> BrowserLoginOutcome
Run one listener-based OAuth login attempt end to end.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
timeout
|
float
|
how long to wait for the redirect once the browser is open. |
180.0
|
ssl_context
|
SSLContext | None
|
TLS for the callback listener ( |
None
|
require_browser
|
bool
|
end the attempt with :class: |
False
|
open_browser
|
Callable[[str], bool] | None
|
replaces |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
A |
BrowserLoginOutcome
|
data: |