daklib.rpc_auth

Authentication and authorization for the DAK RPC server.

Functions

load_tokens(stream)
load_tokens_from_file(token_file)
require_any_scope(context, scopes)	Get the current authentication and verify it has at least one of the given scopes, or abort with PERMISSION_DENIED.
require_auth(context)	Get the current authentication or abort with UNAUTHENTICATED.
require_scope(context, scope)	Get the current authentication and verify it has the given scope, or abort with PERMISSION_DENIED.

Classes

AuthResult(sub, scopes)
AuthenticationInterceptor(token_auth)	gRPC interceptor for authentication
Token(hash, sub, scopes, active, exp)
TokenAuth(tokens)

class daklib.rpc_auth.AuthResult(sub: str, scopes: frozenset[str])

has_any_scope(scopes: Collection[str]) → bool

has_scope(scope: str) → bool

scopes: frozenset[str]

sub: str

class daklib.rpc_auth.AuthenticationInterceptor(token_auth: TokenAuth)

gRPC interceptor for authentication

_abc_impl = <_abc._abc_data object>

intercept_service(continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[_Request, _Response] | None], handler_call_details: HandlerCallDetails) → grpc.RpcMethodHandler[_Request, _Response] | None

Intercepts incoming RPCs before handing them over to a handler.

Args:

continuation: A function that takes a HandlerCallDetails and

proceeds to invoke the next interceptor in the chain, if any, or the RPC handler lookup logic, with the call details passed as an argument, and returns an RpcMethodHandler instance if the RPC is considered serviced, or None otherwise.

handler_call_details: A HandlerCallDetails describing the RPC.

Returns:

An RpcMethodHandler with which the RPC may be serviced if the interceptor chooses to service this RPC, or None otherwise.

class daklib.rpc_auth.Token(hash: str, sub: str, scopes: frozenset[str], active: bool, exp: datetime.date | None)

active: bool

exp: date | None

hash: str

scopes: frozenset[str]

sub: str

class daklib.rpc_auth.TokenAuth(tokens: dict[str, Token])

authenticate(token: str) → AuthResult | None

daklib.rpc_auth._abort_unauthenticated(request: object, context: ServicerContext) → NoReturn

daklib.rpc_auth.load_tokens(stream: BinaryIO) → dict[str, Token]

daklib.rpc_auth.load_tokens_from_file(token_file: str) → dict[str, Token]

daklib.rpc_auth.require_any_scope(context: ServicerContext, scopes: Collection[str]) → AuthResult

Get the current authentication and verify it has at least one of the given scopes, or abort with PERMISSION_DENIED.

daklib.rpc_auth.require_auth(context: ServicerContext) → AuthResult

Get the current authentication or abort with UNAUTHENTICATED.

daklib.rpc_auth.require_scope(context: ServicerContext, scope: str) → AuthResult

Get the current authentication and verify it has the given scope, or abort with PERMISSION_DENIED.