Skip to main content

Authentication & Security

Protolink provides a pluggable, robust Authentication & Security framework designed to secure agent-to-agent and client-to-agent communication. This framework spans both the client side (lazily injecting credentials into outgoing requests) and the server side (extracting and validating credentials before route handlers are invoked).

Whether you are communicating over stateless HTTP or establishing persistent WebSocket connections, ProtoLink can authenticate peers before endpoint handlers run. Built-in authenticators verify credentials; application policy remains responsible for interpreting scopes and deciding authorization.

Overview​

The authentication workflow decouples credentials management from both your core cognitive agent code and the low-level network libraries.

A2A requests

The same HTTP Authenticator protects ProtoLink's native task routes and the A2A JSON-RPC endpoint. The A2A Agent Card remains publicly discoverable and translates the configured card schemes into canonical A2A 1.0 securitySchemes and securityRequirements. Within the A2A adapter, task lookup and mutation are scoped to the authenticated principal and request tenant.

Below is a sequence diagram representing a typical authenticated request cycle:

Core Concepts​

The authentication module revolves around three primary data models and abstractions.

SecurityContext​

The SecurityContext represents an active, authenticated session. It encapsulates details about the authenticated principal, tokens, and expiration times.

from dataclasses import dataclass, field
from typing import Any

from protolink.utils import utc_now

@dataclass
class SecurityContext:
    principal_id: str
    token: str
    expires_at: str | None = None
    issued_at: str = field(default_factory=utc_now)
    metadata: dict[str, Any] = field(default_factory=dict)

    def is_expired(self) -> bool:
        """Check if the context token is past its expiration time."""
        ...

SecurityScheme​

The SecurityScheme outlines the metadata of the authentication mechanism used. It directly corresponds to OpenAPI security schemes, allowing agents to advertise their security protocols in their dynamic agent card metadata.

from dataclasses import dataclass, field
from typing import Any

@dataclass
class SecurityScheme:
    auth_type: str  # e.g., "apiKey", "http", "oauth2"
    auth_scheme: str | None  # e.g., "bearer", "basic" (required if type is "http")
    description: str
    metadata: dict[str, Any] = field(default_factory=dict)

Authenticator (Base Class)​

All security providers inherit from the Authenticator abstract base class. It specifies the API that security handlers must implement to validate credentials.

from abc import ABC, abstractmethod

class Authenticator(ABC):
    @property
    @abstractmethod
    def security_scheme(self) -> SecurityScheme:
        """Expose metadata describing the security scheme."""
        pass

    @abstractmethod
    async def authenticate(self, credentials: str) -> SecurityContext:
        """Authenticate raw credentials and return a SecurityContext."""
        pass

    @abstractmethod
    async def refresh_token(self, context: SecurityContext) -> SecurityContext:
        """Refresh a token, or explicitly return the unchanged context."""
        ...

Built-in Providers​

Protolink includes several security providers out of the box. The local API-key, Basic, and HMAC JWT providers validate credentials in-process. The OAuth delegation provider calls an external token-exchange endpoint and is intentionally a small integration primitive rather than a complete OAuth client.

APIKeyAuth validates simple API keys against a dictionary of known keys. Each value is currently accepted as scope metadata for configuration compatibility, but the authenticator only checks membership; it does not add those scopes to the returned SecurityContext.

from protolink.security.auth import APIKeyAuth

auth = APIKeyAuth(
    valid_keys={
        "sk-12345": ["read", "write"],
        "sk-abcde": ["read"]
    }
)

Server-Side Authentication​

When hosting an agent server, endpoints can be protected by configuring an authenticator on the transport. The transport backend (FastAPI or Starlette) will intercept incoming HTTP requests, extract headers, and invoke the validator.

Authentication and TLS are different layers

An Authenticator decides whether an application request may access the agent. TLSConfig encrypts the network connection and verifies certificate identities before that request arrives. Configure tls= on an HTTP, SSE JSON-RPC, WebSocket, or gRPC transport with its secure URL scheme, then pass that transport to the Agent. Combine TLS with an authenticator when you need both protected traffic and application authorization. See TLS and mutual TLS.

Request Interception Flow​

  1. Public-route check: The A2A Agent Card at /.well-known/agent-card.json plus /healthz and /readyz bypass application authentication. The native /.well-known/agent.json card does not bypass it.
  2. Extraction: For every other route, the server calls the extract_credentials() utility, searching the request in order:
    • Authorization header with Bearer, Basic, or ApiKey prefix.
    • X-API-Key header.
    • Query parameters: api_key, apikey, or token.
  3. Verification: If credentials are found, they are sent to the transport's Authenticator.authenticate(credentials) method.
  4. Rejection: If credentials are missing, or verification raises an exception, the request is terminated immediately, returning an HTTP 401 Unauthorized status code with a JSON error message payload.

Setup Route Protection​

To secure your server-side endpoints, provide the authenticator argument to your HTTPTransport:

from protolink.transport import HTTPTransport
from protolink.security.auth import APIKeyAuth

# Secure HTTP transport using FastAPI backend
transport = HTTPTransport(
    url="http://127.0.0.1:8000",
    backend="fastapi",
    authenticator=APIKeyAuth(valid_keys={"my-secret": ["write"]})
)

WebSocket Handshake Authentication​

WebSocket connections are authenticated during the HTTP connection upgrade handshake phase. The WebSocketTransport intercepts the request headers using a process_request hook before the socket upgrades, ensuring unauthenticated clients are denied connection with a 401 Unauthorized HTTP response immediately.

from protolink.transport import WebSocketTransport
from protolink.security.auth import APIKeyAuth

ws_transport = WebSocketTransport(
    url="ws://127.0.0.1:8080",
    authenticator=APIKeyAuth(valid_keys={"ws-key": ["connect"]})
)

SSE JSON-RPC Authentication​

SSEJSONRPCTransport inherits the HTTP transport's authentication behavior. Unary calls and the long-lived POST /tasks/stream request use the same lazy outbound context and generated headers. On the server, the Starlette or FastAPI backend validates credentials before it opens the event stream, so a rejected request never reaches the streaming handler.

from protolink.security import BearerTokenAuth
from protolink.transport import SSEJSONRPCTransport

sse_transport = SSEJSONRPCTransport(
    url="https://agent.example.com",
    authenticator=BearerTokenAuth(secret="shared-signing-secret"),
    credentials="signed.jwt",
)

gRPC Metadata Authentication​

GRPCTransport applies the same lazy outbound authentication before unary or stream calls, then translates the generated headers into lowercase gRPC metadata. The receiving transport extracts credentials from invocation metadata and aborts the ProtoLink Invoke or Stream RPC with UNAUTHENTICATED when credentials are absent or invalid.

from protolink.security import APIKeyAuth
from protolink.transport import GRPCTransport

grpc_transport = GRPCTransport(
    url="grpcs://agent.example.com:50051",
    authenticator=APIKeyAuth(valid_keys={"service-key": []}),
    credentials="service-key",
)

The optional standard gRPC health and reflection services are registered by gRPC itself and are not passed through ProtoLink's generic invocation authenticator.

Client-Side Authentication​

On the client side, ProtoLink supports Lazy Authentication. When instantiating an HTTP, SSE JSON-RPC, WebSocket, or gRPC transport, provide both an authenticator and its raw credentials string.

from protolink.transport import HTTPTransport
from protolink.security.auth import APIKeyAuth

client_transport = HTTPTransport(
    url="http://127.0.0.1:8000",
    authenticator=APIKeyAuth(valid_keys={"key123": ["read"]}),
    credentials="key123"
)

Automatic Header Injection​

  • Lazy Evaluation: On the first outbound .send() or .subscribe() call supported by the configured HTTP, SSE JSON-RPC, WebSocket, or gRPC transport, it automatically invokes await authenticator.authenticate(credentials).
  • Caching: The resulting SecurityContext is stored in the transport instance for subsequent calls.
  • Signing: Based on the SecurityScheme defined by the authenticator, the transport overrides headers in _build_headers() for outgoing requests:
    • Bearer: Adds Authorization: Bearer <token>
    • Basic: Adds Authorization: Basic <token> without transforming the token. Supply a Base64 payload for standards-compliant HTTP Basic wire format; raw username:password remains useful only when both ProtoLink peers intentionally accept it.
    • ApiKey: Adds X-API-Key: <key> and Authorization: ApiKey <key>

Agent Integration​

The Agent class encapsulates transport management and automatically maps security metadata to the AgentCard.

from protolink.agents import Agent
from protolink.security.auth import BasicAuth

agent = Agent(
    card={"name": "secure-agent", "description": "Needs login", "url": "http://127.0.0.1:8000"},
    transport="http",
    authenticator=BasicAuth(valid_credentials={"admin": "secret"}),
    credentials="admin:secret"
)

Card Security Schemes​

When an agent is initialized with an authenticator, its native metadata card includes the advertised security schemes. With authentication enabled, callers need credentials to read GET /.well-known/agent.json. When the HTTP Agent also uses a2a=True, the public A2A card at GET /.well-known/agent-card.json advertises the translated standard security requirements without requiring credentials.

Example payload for an agent card:

{
  "name": "secure-agent",
  "description": "Needs login",
  "url": "http://127.0.0.1:8000",
  "securitySchemes": {
    "http": {
      "type": "http",
      "scheme": "basic",
      "description": "HTTP Basic authentication (username:password)",
      "metadata": {}
    }
  }
}

Credential Extraction Helper​

You can use the built-in credential extraction logic for custom route middleware or custom transport backends:

from protolink.security.auth import extract_credentials

headers = {"Authorization": "Bearer my-jwt-token"}
query_params = {"api_key": "some-api-key"}

# Returns "my-jwt-token" (headers have precedence)
credentials = extract_credentials(headers, query_params)

Custom Authenticators​

To integrate custom enterprise identity management (e.g. LDAP, active directory, Auth0), subclass Authenticator:

from protolink.security.auth import Authenticator, SecurityScheme, SecurityContext

class LDAPAuthenticator(Authenticator):
    
    @property
    def security_scheme(self) -> SecurityScheme:
        return SecurityScheme(
            auth_type="http",
            auth_scheme="basic",
            description="Active Directory / LDAP validation"
        )

    async def authenticate(self, credentials: str) -> SecurityContext:
        username, password = credentials.split(":")
        # Implement custom LDAP check logic here
        success = my_ldap_library.verify(username, password)
        
        if not success:
            raise ValueError("Invalid LDAP credentials")
            
        return SecurityContext(
            principal_id=username,
            token=credentials
        )

    async def refresh_token(self, context: SecurityContext) -> SecurityContext:
        # LDAP credentials have no token refresh protocol in this example.
        return context

API Reference​

Security moduleAuthentication

The credential verification layer for incoming agent requests, advertised security schemes, outgoing credentials, bearer tokens, and custom authenticators.

protolink.security
SecurityContextSecuritySchemeBearer tokensCustom authenticatorsHTTP integration
ContextCarries the verified principal, token, timestamps, and provider metadata after authentication succeeds.SecurityContext
SchemeDescribes the authentication mechanism advertised on agent cards.SecurityScheme
AuthenticatorDefines authenticate and refresh methods for built-in or application-specific credentials.Authenticator
Bearer JWTVerifies signed bearer tokens with issuer, audience, algorithm, and leeway controls.BearerTokenAuth

Core authentication types​

The data objects in this section are intentionally small. They can be attached to transport requests, request state, policy context, or an AgentCard without bringing a provider SDK into the rest of the application.

SecurityContext​

dataclassprotolink.security.SecurityContext
source
class SecurityContext(
  principal_id: str,
  token: str,
  expires_at: str | None = None,
  issued_at: str = field(default_factory=utc_now),
  metadata: dict[str, Any] = field(default_factory=dict),
)

Represent the authenticated identity produced by an Authenticator. The context carries the credential that was accepted, provider timestamps, and application-specific metadata so later policy and transport layers do not need to authenticate the request again.

Parameters

principal_idstrrequired

Stable identifier for the authenticated user, service, client, or agent. ProtoLink does not interpret its format; use a value that your policy and persistence layers can consistently recognize.

tokenstrrequired

The accepted credential or delegated access token. This value is retained verbatim and is included by to_dict(), so do not log or serialize the context into an untrusted destination.

expires_atstr | Nonedefault: None

Optional absolute expiration timestamp in ISO 8601 format. Use a timezone-aware value such as 2026-07-20T12:30:00+00:00 so it can be compared with ProtoLink's UTC clock.

issued_atstrdefault: utc_now()

ISO timestamp describing when the token was issued. When omitted, each new instance receives the current UTC time from utc_now().

metadatadict[str, Any]default: {}

Provider- or application-specific details, such as verified custom JWT claims. A new dictionary is created for every context.

Security boundary

A SecurityContext records the result of authentication; creating one directly does not verify a token. Only trust contexts returned by an authenticator or another trusted boundary.

SecurityContext.is_expired​

methodprotolink.security.SecurityContext.is_expired
source
is_expired() -> bool

Compare the context's absolute expiration timestamp with the current UTC time. This is a local timestamp check; it does not contact the issuing provider, inspect revocation state, or refresh the credential.

Returns

expiredbool

True when the current UTC time is later than expires_at. Returns False when no expiration was supplied.

Raises

ValueError

Raised by datetime.fromisoformat() when expires_at is not a valid ISO timestamp.

TypeError

Raised when a timezone-naive timestamp is compared with ProtoLink's timezone-aware UTC clock.

SecurityContext.to_dict​

methodprotolink.security.SecurityContext.to_dict
source
to_dict() -> dict

Create the dictionary representation used by serializers and integration code. All five context fields are included, including the raw token.

Returns

contextdict[str, Any]

Mapping with principal_id, token, expires_at, issued_at, and metadata keys. The outer mapping is new, but the metadata dictionary is not deep-copied.

SecurityScheme​

dataclassprotolink.security.auth.SecurityScheme
source
class SecurityScheme(
  auth_type: SecuritySchemeType,
  auth_scheme: HttpAuthScheme | None,
  description: str,
  metadata: dict[str, Any] = field(default_factory=dict),
)

Describe the authentication mechanism advertised by an authenticator. Agent metadata and transport code use this provider-neutral value to decide how a credential should be represented on the wire.

Parameters

auth_type"apiKey" | "http" | "oauth2" | "mutualTLS" | "openIdConnect"required

Broad OpenAPI-style category for the authentication mechanism.

auth_schemeHttpAuthScheme | Nonerequired

HTTP authentication scheme such as "bearer" or "basic". Pass None for non-HTTP scheme types. The annotation also accepts digest, hmac, negotiate, ntlm, aws4auth, hawk, and edgegrid.

descriptionstrrequired

Human-readable explanation suitable for discovery metadata.

metadatadict[str, Any]default: {}

Scheme-specific extensions. Built-in bearer schemes report supported algorithms plus configured issuer and audience; OAuth delegation reports its exchange endpoint.

Import path

SecurityScheme is defined in protolink.security.auth but is not currently re-exported from protolink.security. Custom authenticators should import it from the defining module.

SecurityScheme.to_dict​

methodprotolink.security.auth.SecurityScheme.to_dict
source
to_dict() -> dict

Convert the scheme into the wire-oriented field names used in discovery metadata.

Returns

schemedict[str, Any]

Mapping with type, scheme, description, and metadata keys. The metadata value is shared with the dataclass rather than deep-copied.

Authenticator contract​

All built-in and application-defined authenticators implement the same asynchronous validation boundary. The transport owns when the methods are called; the authenticator owns how a raw credential becomes a trusted context.

Authenticator​

abstract classprotolink.security.Authenticator
source
class Authenticator

Abstract base class for credential providers. Implementations must advertise a SecurityScheme, validate raw credentials asynchronously, and provide an explicit refresh behavior even when refresh is a no-op.

Abstract members

security_schemeSecurityScheme

Read-only property describing how the provider is advertised and how a client transport should present its credential.

authenticateasync (str) -> SecurityContext

Required credential-validation method.

refresh_tokenasync (SecurityContext) -> SecurityContext

Required refresh hook. Providers that cannot refresh return the original context unchanged.

Subclass requirement

All three members are abstract. A custom subclass remains non-instantiable until it implements security_scheme, authenticate(), and refresh_token().

Authenticator.security_scheme​

abstract propertyprotolink.security.Authenticator.security_scheme
source
security_scheme -> SecurityScheme

Return a declarative description of the provider. Transports inspect this value to construct outbound headers, while agents expose it through discovery metadata.

Returns

schemeSecurityScheme

Provider category, optional HTTP scheme, human-readable description, and any provider metadata.

Authenticator.authenticate​

abstract async methodprotolink.security.Authenticator.authenticate
source
await authenticate(
  credentials: str,
) -> SecurityContext

Validate raw credentials and translate them into the common authenticated principal context. Implementations may perform local cryptographic or dictionary checks, or await an external identity provider.

Parameters

credentialsstrrequired

Credential payload after transport-level prefix removal. Its expected syntax depends on the concrete provider.

Returns

contextSecurityContext

Verified principal, accepted token, timestamps, and optional metadata.

Raises

authentication error

Concrete providers raise when credentials are missing, malformed, unverifiable, expired, or rejected by an external identity provider. The base interface does not define a specialized exception type.

Authenticator.refresh_token​

abstract async methodprotolink.security.Authenticator.refresh_token
source
await refresh_token(
  context: SecurityContext,
) -> SecurityContext

Refresh an authenticated context when the provider supports renewable credentials. This hook is part of the provider contract, but ProtoLink does not schedule refresh automatically.

Parameters

contextSecurityContextrequired

Existing authenticated context whose token should be renewed or retained.

Returns

contextSecurityContext

Refreshed context, or the original object for providers whose refresh implementation is a no-op.

Built-in provider reference​

BearerTokenAuth​

classprotolink.security.BearerTokenAuth
source
class BearerTokenAuth(
  secret: str,
  algorithm: str = "HS256",
  *,
  issuer: str | None = None,
  audience: str | None = None,
  leeway_seconds: int = 0,
)

Validate compact JWTs signed with a shared HMAC secret. The implementation is dependency-free and deliberately restricted to symmetric HS256, HS384, and HS512 signatures; it does not fetch JWK sets or accept asymmetric algorithms.

Parameters

secretstrrequired

Non-empty shared signing secret used to recompute and constant-time compare the JWT signature.

algorithm"HS256" | "HS384" | "HS512"default: "HS256"

Exact algorithm required in the JWT header and used for HMAC verification. Algorithm substitution is rejected.

issuerstr | Nonedefault: None

When set, require the payload's iss claim to equal this value.

audiencestr | Nonedefault: None

When set, require this value in the payload's string or string-list aud claim.

leeway_secondsintdefault: 0

Non-negative clock-skew allowance applied to exp, nbf, and iat validation.

Raises

ValueError

Raised immediately for an empty secret, unsupported algorithm, or negative leeway.

Advertised scheme

security_schemeSecurityScheme

HTTP bearer scheme whose metadata lists all three supported HMAC algorithms and the configured issuer and audience.

BearerTokenAuth.authenticate​

async methodprotolink.security.BearerTokenAuth.authenticate
source
await authenticate(
  credentials: str,
) -> SecurityContext

Decode and verify one compact JWT, validate its registered claims, and build the corresponding principal context.

Parameters

credentialsstrrequired

Three-segment compact JWT without the Bearer prefix. Header and payload segments must be base64url-encoded JSON objects.

Returns

contextSecurityContext

Context whose principal is sub, then client_id, then "unknown". Verified exp and iat NumericDate claims become UTC ISO timestamps. A dictionary-valued metadata claim is retained; other non-registered claims are nested under metadata["claims"].

Raises

Exception

Any malformed segment, invalid JSON, algorithm mismatch, signature mismatch, invalid time claim, expiry, premature validity, future issue time, issuer mismatch, or audience mismatch is wrapped as Exception("Token authentication failed: …").

Claim validation

The provider validates exp, nbf, iat, and optionally iss and aud. It does not require a subject, consult a revocation list, or enforce application-specific authorization claims.

BearerTokenAuth.refresh_token​

async methodprotolink.security.BearerTokenAuth.refresh_token
source
await refresh_token(
  context: SecurityContext,
) -> SecurityContext

Return the existing bearer context unchanged. A signed access token cannot be renewed without a separate issuer or refresh-token protocol, which this local validator does not implement.

Parameters

contextSecurityContextrequired

Existing bearer context.

Returns

contextSecurityContext

The exact same object passed by the caller.

OAuth2DelegationAuth​

classprotolink.security.OAuth2DelegationAuth
source
class OAuth2DelegationAuth(
  exchange_endpoint: str,
  client_id: str,
  client_secret: str,
)

Exchange an incoming subject token for a narrower access token at an external HTTP endpoint. The provider uses httpx.AsyncClient and sends a compact JSON token-exchange request.

Parameters

exchange_endpointstrrequired

URL receiving the token-exchange POST request.

client_idstrrequired

OAuth client identifier included in the JSON request body.

client_secretstrrequired

OAuth client secret included in the JSON request body. Constructor values are stored without validation.

Advertised scheme

security_schemeSecurityScheme

OAuth 2 scheme whose metadata exposes the configured exchange endpoint.

OAuth2DelegationAuth.authenticate​

async methodprotolink.security.OAuth2DelegationAuth.authenticate
source
await authenticate(
  credentials: str,
) -> SecurityContext

POST a subject-token exchange request and translate a successful JSON response into a SecurityContext.

Parameters

credentialsstrrequired

Broad-scoped subject token sent as subject_token. The provider also sends the standard token-exchange grant type plus its client ID and secret.

Returns

contextSecurityContext

Context populated from response fields: sub defaults to "unknown", access_token defaults to an empty string, and metadata defaults to an empty dictionary.

Raises

Exception

Non-200 responses, network failures, JSON decoding errors, and response conversion failures are wrapped as Exception("OAuth delegation failed: …").

dependency error

A missing httpx installation is also caught and wrapped in the same generic OAuth-delegation Exception.

Current expiration behavior

The current implementation copies the response's expires_in value directly into SecurityContext.expires_at. OAuth servers usually return a relative number of seconds there, while SecurityContext.is_expired() expects an absolute ISO timestamp. Normalize that field in an application-specific provider before calling is_expired().

OAuth2DelegationAuth.refresh_token​

async methodprotolink.security.OAuth2DelegationAuth.refresh_token
source
await refresh_token(
  context: SecurityContext,
) -> SecurityContext

Return the delegated context unchanged. Despite the method name, the current provider does not call a refresh endpoint or retain an OAuth refresh token.

Parameters

contextSecurityContextrequired

Existing delegated context.

Returns

contextSecurityContext

The exact same object passed by the caller.

APIKeyAuth​

classprotolink.security.APIKeyAuth
source
class APIKeyAuth(
  valid_keys: dict[str, list[str]],
)

Perform an in-memory membership check for static service credentials. This provider is useful for small service-to-service deployments where keys are loaded and rotated by the application.

Parameters

valid_keysdict[str, list[str]]required

Mapping from accepted credential strings to scope lists. The mapping is stored by reference. The current implementation checks only whether the key exists; it does not inspect or propagate the associated list.

Advertised scheme

security_schemeSecurityScheme

API-key scheme with no HTTP sub-scheme and no provider metadata.

Credential storage

Keys and dictionary lookups are plain Python strings. Hash keys at the application boundary or use a constant-time external verifier when your threat model requires protection against memory disclosure or timing analysis.

APIKeyAuth.authenticate​

async methodprotolink.security.APIKeyAuth.authenticate
source
await authenticate(
  credentials: str,
) -> SecurityContext

Check whether the credential is a key in the configured mapping.

Parameters

credentialsstrrequired

Raw API key after transport-level prefix extraction.

Returns

contextSecurityContext

Non-expiring context whose principal ID is "api-key-" followed by the first eight key characters and whose token is the complete key. The configured scope list is not copied into metadata.

Raises

Exception

Raised with "Invalid API key" when the credential is absent from valid_keys.

APIKeyAuth.refresh_token​

async methodprotolink.security.APIKeyAuth.refresh_token
source
await refresh_token(
  context: SecurityContext,
) -> SecurityContext

Return the static-key context unchanged. API keys do not have a refresh protocol; rotate the configured key mapping instead.

Parameters

contextSecurityContextrequired

Existing API-key context.

Returns

contextSecurityContext

The exact same object passed by the caller.

BasicAuth​

classprotolink.security.BasicAuth
source
class BasicAuth(
  valid_credentials: dict[str, str],
)

Validate HTTP Basic-style username and password pairs against an in-memory mapping. The authenticator accepts either the decoded username:password text or the Base64 payload normally carried after an Authorization: Basic prefix.

Parameters

valid_credentialsdict[str, str]required

Mapping from usernames to expected plaintext passwords. The mapping is stored by reference and constructor values are not validated.

Advertised scheme

security_schemeSecurityScheme

HTTP Basic scheme used by transport header construction and discovery.

Use with TLS

Basic authentication only encodes credentials; it does not encrypt them. Always use HTTPS or another confidential transport outside local development.

BasicAuth.authenticate​

async methodprotolink.security.BasicAuth.authenticate
source
await authenticate(
  credentials: str,
) -> SecurityContext

Decode the credential when it is valid Base64, split the resulting text at the first colon, and compare the username/password pair with the configured mapping.

Parameters

credentialsstrrequired

Base64-encoded or raw username:password string. Passwords may contain additional colons because splitting occurs only once.

Returns

contextSecurityContext

Non-expiring context whose principal ID is the decoded username and whose token remains the original encoded or raw credential string.

Raises

Exception

Raised for a decoded value without a colon or for a username/password pair that does not exactly match the mapping.

Decoding order

The method attempts strict Base64 decoding first and treats the original value as raw text only when decoding fails. A raw value that also happens to be valid Base64 is therefore interpreted as encoded credentials.

BasicAuth.refresh_token​

async methodprotolink.security.BasicAuth.refresh_token
source
await refresh_token(
  context: SecurityContext,
) -> SecurityContext

Return the Basic-authentication context unchanged. Rotate passwords in the application's credential store rather than through this hook.

Parameters

contextSecurityContextrequired

Existing Basic-authentication context.

Returns

contextSecurityContext

The exact same object passed by the caller.

Credential helper reference​

extract_credentials​

functionprotolink.security.extract_credentials
source
extract_credentials(
  headers: Any,
  query_params: dict[str, str] | None = None,
) -> str | None

Extract one raw credential from HTTP-style headers or query parameters using a fixed precedence order. The helper removes known authorization prefixes but does not authenticate or decode the returned value.

Parameters

headersAnyrequired

Header collection. Mapping-like objects are read with .get(); other values are expected to iterate as (name, value) pairs. Header name handling covers canonical and lowercase names for mappings and is case-insensitive for pair iterables.

query_paramsdict[str, str] | Nonedefault: None

Optional query mapping checked only when no usable authorization or API key header was found.

Returns

credentialsstr | None

First credential selected by precedence after surrounding whitespace and a recognized authorization prefix are removed, or None when no supported location is present. A prefix-only or whitespace-only value can produce the empty string.

Precedence

1Authorization header

Removes a case-insensitive Bearer , Basic , ApiKey , or apikey prefix. Unknown schemes are returned as raw stripped header text.

2X-API-Key header

Uses the stripped header value when no authorization credential won.

3query parameter

Checks api_key, then apikey, then token.

Raises

collection error

The helper does not normalize arbitrary header objects. Invalid iterable shapes, non-string names or values, and custom .get() failures may propagate their native exceptions.

Examples

from protolink.security import extract_credentials

extract_credentials({"Authorization": "Bearer signed.jwt"})
# "signed.jwt"

extract_credentials({}, {"token": "query-token"})
# "query-token"