Authentication API

The OAuth2 layer: token models, the store protocol and its backends, the headless code-exchange/refresh functions, the provider that clients consume, and the login callback listener.

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])

JSON-file token store, written atomically with 0o600 permissions.

load

load() -> TokenSet | None

The stored token set, or None when no store file exists.

Raises:

Type	Description
AnafConfigError	the file exists but cannot be read or parsed — stay in the AnafError hierarchy instead of leaking a raw pydantic/OS error.

clear

clear() -> None

Delete the store file; a no-op when none exists.

Raises:

Type	Description
AnafConfigError	the file exists but cannot be removed.

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).

exchange_code async

exchange_code(http: AsyncClient, *, client_id: str, client_secret: str, code: str, redirect_uri: str) -> TokenSet

Exchange an authorization code for a token set (no certificate needed).

refresh_tokens async

refresh_tokens(http: AsyncClient, *, client_id: str, client_secret: str, refresh_token: str) -> TokenSet

Get a fresh token set from a refresh token (headless; rotates refresh token).

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 httpx 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

httpx 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 an ssl.SSLContext to 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.