Auth & security¶
Authentication and the security-agreement machinery: SIP digest and IMS-AKA challenges, P-CSCF IPsec sec-agree (3GPP TS 33.203 / RFC 3329), and STIR/SHAKEN signing and verification.
from siphon import auth
@proxy.on_request("INVITE")
async def route(request):
if not await auth.verify_digest(request, "example.com"):
await auth.require_proxy_digest(request, "example.com")
return
request.relay()
Challenging in B2BUA mode¶
The digest helpers take a Request or a Call. This matters: registering
any @b2bua.* handler makes the dispatcher route INVITE straight to the B2BUA
path, so @proxy.on_request never sees it and a proxy-style challenge would
simply never run.
from siphon import auth, b2bua, log
@b2bua.on_invite
async def new_call(call):
if not await auth.require_proxy_digest(call, realm="example.com"):
return # 407 armed; siphon answers the A-leg
log.info(f"call from {call.auth_user}")
call.dial(str(call.ruri))
Returning False arms the challenge as the call's deferred reject, the same one
call.reject() produces — siphon answers the A-leg INVITE and drops the call
actor, so no B-leg is dialled for an unauthenticated caller. On success the
caller's Proxy-Authorization is stripped from the message the B-leg INVITE is
built from, because it is hop-by-hop (RFC 3261 §22.3); forwarding it would only
make the next hop challenge credentials that were minted for us. The verified
username lands on call.auth_user and on the call's CDR.
This is the opposite direction from call.dial(auth_passthrough=True), where a
downstream PBX issues the challenge and siphon relays it end-to-end for the
caller to answer. Use auth_passthrough when the credentials live at the far
end; challenge on the Call when siphon owns them.
When the credential is not the subscriber identity¶
request.auth_user / call.auth_user hold the username exactly as it appeared
in the Authorization / Proxy-Authorization header, because that is the
string the digest response was computed over — siphon must not guess at its
structure.
Deployments where the authentication identity is not the subscriber identity
need to say so. IMS is the standard case: a private identity user@realm
authenticates a public identity. Any scheme carrying a validity prefix or a
tenant qualifier in the username has the same shape. Both properties are
writable, so the script reduces the credential after verification:
@proxy.on_request("REGISTER")
async def register(request):
if not await auth.verify_digest(request, realm="example.com"):
await auth.require_www_digest(request, realm="example.com")
return
request.auth_user = request.auth_user.split(":", 1)[1]
registrar.save(request)
Everything keyed on the authenticated identity reads the new value:
registrar.enforce_auth_aor_match, which compares it to the AoR userpart, and
the CDR's auth_user field. That AoR check is exactly why the reduction has to
happen — an unreduced credential never equals the AoR userpart, so every
REGISTER is answered 403 and the only way to deploy is to turn the anti-hijack
check off entirely.
Assigning it before verifying defeats that comparison. It is an assertion about an identity already proven, not a way to prove one — so set it only on the success path.
require_ims_digest and require_aka_digest take a Request only — IMS and
AKA digest are REGISTER-time procedures, and REGISTER never reaches the B2BUA
path.
auth namespace¶
Which algorithms a challenge offers¶
require_www_digest / require_proxy_digest emit one WWW-Authenticate /
Proxy-Authenticate header per algorithm in auth.algorithms, in that order,
all sharing one nonce (RFC 7616 §3.7 — weakest first, so a legacy MD5-only
client finds its entry and a modern one takes the strongest it supports).
Narrow it for a client population that cannot take a multi-challenge 401. The
RFC's SHOULD assumes a client picks one from the list; some SDKs instead
abandon the registration when a 401 carries more than one WWW-Authenticate,
whatever the algorithms are and whatever the order, and there is nothing in the
exchange to fall back to:
You pay for that in the algorithms you drop, so narrow only as far as the clients require. Verification is unaffected either way — siphon accepts any algorithm it can compute, including ones it did not offer.
An unknown name fails the config load rather than being skipped: a silently
dropped entry is a challenge set the operator did not choose, and the
difference does not show up until clients stop registering. An empty list is
refused for the same reason — a 401 carrying no challenge is unanswerable.
AKAv1-MD5 is refused here too: IMS AKA is network-selected, and its challenge
(with the AKA nonce and ck=/ik=) comes from auth.require_aka_digest() or
auth.require_ims_digest().
Issuing your own challenge¶
require_www_digest / require_proxy_digest build the challenge for you. A
script that verifies credentials itself — rather than through a configured
auth.backend — builds its own WWW-Authenticate header instead, and needs the
engine's nonce for it:
@proxy.on_request("REGISTER")
def register(request):
header = request.get_header("Authorization")
if header is None:
nonce = auth.generate_nonce()
request.set_reply_header(
"WWW-Authenticate",
f'Digest realm="{realm}", nonce="{nonce}", algorithm=MD5, qop="auth"',
)
request.reply(401, "Unauthorized")
return
if not auth.validate_nonce(nonce_of(header)):
return challenge(request) # stale — re-challenge, do not trust it
...
auth.generate_nonce() mints {unix_seconds:016x}.{tag} — the timestamp is
embedded rather than stored, so any instance in a fleet can reject a stale nonce
without shared state. auth.validate_nonce(nonce) returns True only for a nonce
this engine minted, no older than auth.nonce_ttl_secs, not future-dated beyond
60 s of clock skew, and carrying a matching HMAC tag when auth.nonce_secret is
configured.
Validating the nonce is what bounds replay. Without it a captured
Authorization is replayable forever, which is why the built-in verify_digest
/ require_*_digest paths always check it — this pair is for scripts that do
not go through them.
Verifying against a credential you hold¶
By default the digest helpers verify against the configured credential source
(auth.backend). Pass password= or ha1= (not both) to verify against a
credential the script supplies instead, which short-circuits the backend lookup
entirely — a deployment that derives credentials in-process then needs no
credential source configured at all, rather than standing up an HTTP endpoint
for siphon to fetch a value the script already has.
@proxy.on_request("REGISTER")
async def register(request):
secret = await cache.fetch("secrets", derive_key(request))
if secret is None or not await auth.verify_digest(request, realm, password=secret):
await auth.require_www_digest(request, realm)
return
registrar.save(request)
Accepted by verify_digest, require_digest, require_www_digest and
require_proxy_digest, on a Request or a Call alike.
| kwarg | Holds | Trade-off |
|---|---|---|
password= |
the plaintext secret | One secret answers MD5, SHA-256 and SHA-512-256, because H(A1) is derived with whatever algorithm the client actually used (RFC 7616 §3.4.3). |
ha1= |
an already-computed H(A1) | The deployment never stores plaintext, but the hash is bound to the one algorithm it was computed for — a client answering with another will not verify. |
Passing both raises ValueError: one of them would be silently ignored and the
script author could not tell which.
Everything else is unchanged by a supplied credential. The anti-replay nonce
check still runs, so a captured Authorization cannot be replayed (RFC 7616
§3.3). A rejection still arms the 401/407 rather than returning a silent
False, still counts toward failed_auth_ban, and still increments
siphon_credential_failures_total — a path that authenticates without counting
would be a blind spot for anyone alerting on brute force.
A rejection is the case above: credentials were presented and refused. Two neighbouring cases score differently, and both used to be conflated with it:
- No credentials at all. The RFC 3261 §22.2 opening leg. Counted at
failed_auth_ban.missing_credentials_weight, which is0by default, and visible insiphon_auth_failures_total. - The credential source could not answer — an HTTP auth backend timeout or
connection failure, or no usable backend configured. Never counted: it says
nothing about the peer. It increments
siphon_auth_backend_errors_total, which is worth an alert of its own — a non-zero rate means authentication is failing into 401s for every subscriber.
See Hardening & security for the full scoring table.
Mock authentication namespace.
Control auth behavior in tests::
from siphon import auth
auth._allow = True # all auth checks pass
auth._allow = False # all auth checks fail (challenge sent)
generate_nonce
¶
Mint a digest challenge nonce.
Shape is {unix_seconds:016x}.{tag}, matching the engine, which
embeds the timestamp rather than storing it so any instance in a fleet
can reject a stale nonce without shared state.
A script that builds its own WWW-Authenticate /
Proxy-Authenticate header — because it verifies credentials itself
rather than through a configured backend — mints the nonce here::
nonce = auth.generate_nonce()
request.set_reply_header(
"WWW-Authenticate",
f'Digest realm="{realm}", nonce="{nonce}", algorithm=MD5, qop="auth"',
)
request.reply(401, "Unauthorized")
The mock uses a random tag (the engine does too when no
auth.nonce_secret is configured); it does not compute the HMAC
variant, so a mock-minted nonce is not interchangeable with a real one.
validate_nonce
¶
Whether nonce is well-formed and still fresh.
Mirrors the engine's freshness rules: a well-formed
{timestamp}.{tag}, no older than the TTL, and not implausibly
future-dated (60 s of clock skew allowed). This is what bounds replay
of a captured Authorization for a script doing its own digest
check::
if not auth.validate_nonce(nonce_from_the_header):
await auth.require_www_digest(request, realm) # stale — re-challenge
return
The mock checks freshness only — it has no shared secret, so it cannot verify the engine's HMAC tag.
add_user
¶
Add credentials for testing (test helper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
realm
|
str
|
Auth realm (e.g. |
required |
username
|
str
|
Username. |
required |
password
|
str
|
Password. |
required |
require_www_digest
¶
require_www_digest(
target: Any,
realm: Optional[str] = None,
password: Optional[str] = None,
ha1: Optional[str] = None,
) -> bool
Challenge with 401 WWW-Authenticate, or verify existing credentials.
If credentials are valid: sets target.auth_user, returns True.
Otherwise: arms a 401 response, returns False.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target
|
Any
|
The SIP |
required |
realm
|
Optional[str]
|
Auth realm (e.g. |
None
|
password
|
Optional[str]
|
Verify against this plaintext secret instead of the
configured backend. Mutually exclusive with |
None
|
ha1
|
Optional[str]
|
Verify against this already-computed H(A1) instead of the
configured backend. Mutually exclusive with |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
|
require_proxy_digest
¶
require_proxy_digest(
target: Any,
realm: Optional[str] = None,
password: Optional[str] = None,
ha1: Optional[str] = None,
) -> bool
Challenge with 407 Proxy-Authenticate.
Same as :meth:require_www_digest but uses 407. This is the challenge
an INVITE normally gets, including from a B2BUA authenticating its own
A-leg::
@b2bua.on_invite
async def new_call(call):
if not await auth.require_proxy_digest(call, realm="example.com"):
return # 407 armed; siphon answers the A-leg
call.dial(call.ruri)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target
|
Any
|
The SIP |
required |
realm
|
Optional[str]
|
Auth realm. |
None
|
password
|
Optional[str]
|
Verify against this plaintext secret instead of the
configured backend. Mutually exclusive with |
None
|
ha1
|
Optional[str]
|
Verify against this already-computed H(A1) instead of the
configured backend. Mutually exclusive with |
None
|
require_digest
¶
require_digest(
target: Any,
realm: Optional[str] = None,
password: Optional[str] = None,
ha1: Optional[str] = None,
) -> bool
Convenience alias for :meth:require_www_digest.
require_ims_digest
¶
IMS digest authentication via Diameter Cx MAR/MAA.
Sends a Multimedia-Auth-Request to the HSS and uses the returned authentication vector to challenge or verify the UE.
Takes a Request only — unlike :meth:require_www_digest /
:meth:require_proxy_digest, IMS and AKA digest are REGISTER-time
procedures and REGISTER never reaches the B2BUA path, so the engine
does not accept a Call here either.
Returns:
| Type | Description |
|---|---|
bool
|
|
require_aka_digest
¶
IMS AKA digest authentication using local Milenage credentials.
Uses locally-configured K/OP/AMF credentials (from auth.aka_credentials
in siphon.yaml) to generate AKA authentication vectors — no Diameter HSS
connection needed. The nonce contains base64(RAND || AUTN) per 3GPP TS 33.203.
The 401 carries ck= and ik= in WWW-Authenticate, exactly as
the HSS path does, so a P-CSCF in front can set up the IPsec SAs. That
P-CSCF must strip them with reply.take_av() before relaying the 401
to the UE.
Example::
if not auth.require_aka_digest(request, realm="ims.test"):
log.info("sent 401 AKA challenge")
return
Takes a Request only, for the same reason as
:meth:require_ims_digest.
Returns:
| Type | Description |
|---|---|
bool
|
|
stamp_integrity_protected
¶
P-CSCF: stamp integrity-protected into every Authorization header.
3GPP TS 24.229 has the P-CSCF tell the S-CSCF whether a REGISTER came
in over the IPsec SA it set up with the UE, by writing
integrity-protected into the Authorization header. The S-CSCF
then accepts a protected re-/de-REGISTER of a registered user without
a new AKA challenge (:meth:verify_integrity_protected). So the
P-CSCF writes it on every REGISTER, overwriting whatever the UE sent:
a "yes" the UE forged is replaced, never passed through.
Each header gets "yes" only when the request arrived over an SA
(:attr:Request.matched_sa) and that SA was negotiated for the
header's username, i.e. the private identity (IMPI) of the
REGISTER whose 401 keyed the SA. Anything else gets "no". The
IMPI check is what stops a UE holding a valid SA of its own from
claiming protection under another subscriber's IMPI. Every other
parameter of the header is kept as it was.
Call it on every REGISTER, before relaying it::
@proxy.on_request("REGISTER")
def handle_register(request):
...
auth.stamp_integrity_protected(request)
request.relay()
A REGISTER with no Authorization header is left alone.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
The REGISTER |
required |
Returns:
| Type | Description |
|---|---|
Optional[str]
|
|
Optional[str]
|
header, or |
verify_integrity_protected
¶
S-CSCF: accept a protected re-/de-REGISTER without challenging it again.
The counterpart of :meth:stamp_integrity_protected. 3GPP TS 24.229
lets the S-CSCF skip the AKA challenge on a re-REGISTER or
de-REGISTER the P-CSCF received over the UE's IPsec SA. Returns
True only when both of these hold:
- the first
Authorizationheader carryingintegrity-protectedsays"yes","tls-yes"or"ip-assoc-yes", and - that header's
usernameis the identity that authenticated a live binding of theToAoR (:attr:Contact.auth_user, with implicit-set aliases resolved the wayregistrar.lookupdoes).
The second check is what stops a UE with its own SA and its own IMPI from re-registering or de-registering someone else's public identity. The first is only as good as the P-CSCF in front, so use this behind a P-CSCF that always stamps the header.
On True it sets request.auth_user to that username, so the rest
of the handler reads it exactly as after a digest check. On False
nothing changes, and the handler falls back to the challenge::
@proxy.on_request("REGISTER")
def handle_register(request):
if not auth.verify_integrity_protected(request):
if not auth.require_aka_digest(request, realm=REALM):
return
registrar.save(request)
An initial REGISTER never passes: there is no binding for its IMPI yet, so it is challenged. Neither does one whose binding was saved without an authenticated user.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
The REGISTER |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
verify_digest
¶
verify_digest(
target: Any,
realm: Optional[str] = None,
password: Optional[str] = None,
ha1: Optional[str] = None,
) -> bool
Verify credentials without sending a challenge.
By default the credential comes from the configured backend. Pass
password= or ha1= (not both) to verify against a credential the
script supplies instead, which short-circuits the backend lookup
entirely — a deployment that derives credentials in-process then needs
no credential source configured at all::
secret = await cache.fetch("secrets", request.auth_user)
if not await auth.verify_digest(request, realm, password=secret):
await auth.require_www_digest(request, realm)
return
ha1= takes an already-computed H(A1) verbatim, so a deployment can
hold the hash rather than the plaintext. It is algorithm-specific by
construction: a client answering with an algorithm other than the one
the hash was computed for will not verify, where password= covers
MD5, SHA-256 and SHA-512-256 from one secret, because H(A1) is derived
with whatever algorithm the client actually used (RFC 7616 §3.4.3).
The anti-replay nonce check runs either way in the engine: a supplied
credential changes where the secret comes from, never whether a captured
Authorization may be replayed.
What the mock does and does not do. With password= or ha1=
it performs the real RFC 7616 arithmetic — H(A1), H(A2) and the response
under the algorithm the Authorization header names — so a wrong
password returns False and a right one returns True, and a test
that delegates verification to the engine keeps its accept/reject
assertions. With neither kwarg it still answers from the preset
_allow flag, because there is no credential source in the mock to
look one up from.
It deliberately does not replay-check the nonce, which the engine
does first: a fixture's hand-written nonce would otherwise fail every
test for the wrong reason. validate_nonce covers that separately.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target
|
Any
|
The SIP |
required |
realm
|
Optional[str]
|
Auth realm. When |
None
|
password
|
Optional[str]
|
Plaintext secret to verify against. Mutually exclusive
with |
None
|
ha1
|
Optional[str]
|
Already-computed H(A1) to verify against. Mutually exclusive
with |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
|
Raises:
| Type | Description |
|---|---|
ValueError
|
if both |
ipsec namespace¶
P-CSCF IPsec security association management for the sec-agree handshake.
Mock :class:Ipsec namespace.
set_allocate_failure
¶
set_allocate_failure(
exc_type: Optional[type[BaseException]],
message: str = "mock allocate failure",
) -> None
Configure the next allocate() call to raise exc_type.
Pass None to clear and let allocate succeed again.
set_pcscf_families
¶
Model which address families this P-CSCF has a listener for.
Both default to available. Set one to False to model a
single-stack P-CSCF so allocate() raises for a UE of the missing
family, exactly as the Rust binding does.
SecurityOffer¶
A Security-Client offer parsed from a REGISTER (request.parse_security_client()).
Mock :class:SecurityOffer — UE-side IPsec proposal.
Transform¶
An operator-policy transform choice (Transform.HmacSha1_96Null, …).
Mock :class:Transform enum — operator policy choice.
hmac-sha-1-96 and hmac-md5-96 are the 3GPP TS 33.203 Annex H
transforms. hmac-sha-256-128 is a siphon extension: the transform is
RFC 4868, but its 256-bit key comes from siphon's own expansion, so it
interoperates only siphon-to-siphon.
AuthVectorHandle¶
The opaque CK/IK container produced by reply.take_av().
Mock :class:AuthVectorHandle — opaque CK/IK container.
The bytes are not exposed to Python in the real binding; the mock
keeps them accessible via _ck/_ik for tests, but treats them
as consumed after one allocate.
PendingSA¶
An allocated-but-not-yet-active SA pair, returned by ipsec.allocate(...).
Mock :class:PendingSA.
activate
¶
Mark the SA pair active.
hard_lifetime_secs (optional) re-pins the kernel
hard-lifetime on all four SAs of the pair via XFRM_MSG_UPDSA,
without rekeying or disturbing selectors / SPIs. Use on the
path that processes the 200 OK to the auth REGISTER to tighten
the SA expiry from the placeholder value installed at
allocation time (typically the UE's Expires: ask) to the
actual grant from the registrar of record (3GPP TS 33.203 §7.4
— IPsec SA lifetime tracks SIP registration lifetime).
None (default) preserves the original metadata-only
transition.
In the mock, this only updates self.expires_secs so tests
can assert the script wired the grant through correctly.
SecurityServerParams¶
The Security-Server parameters to echo back to the UE.
Mock :class:SecurityServerParams.
SAHandle¶
A read-only view of the active SA that decrypted a request
(request.matched_sa).
Mock :class:SAHandle — read-only view of an active SA returned by
request.matched_sa. Tests can construct one directly and assign
it to request._matched_sa.
impi is a test double for what the engine keeps on the SA but does not
expose to scripts: the private identity (the Authorization username of
the REGISTER whose challenge keyed the SA). :meth:MockAuth.stamp_integrity_protected
reads it the way the engine does; None never counts as a match.
stir namespace¶
STIR/SHAKEN Identity-header signing and verification.
Mock stir namespace — STIR/SHAKEN signing and verification.
Scripts use::
from siphon import stir
@proxy.on_request("INVITE")
def on_invite(request):
origid = stir.sign(request, attestation="A") # add Identity header
request.relay()
@proxy.on_request("INVITE")
def verify_inbound(request):
result = stir.verify(request)
if result.verstat == "TN-Validation-Failed":
request.reply(438, "Invalid Identity Header")
return
stir.apply_verstat(request, result)
request.relay()
Test helpers: set :attr:signing_enabled / :attr:verification_enabled
to simulate config; call :meth:set_verify_result to pin the next
:meth:verify outcome; inspect :attr:signed / :attr:applied_verstats.
sign
¶
sign(
request: Any,
attestation: str = "A",
origid: Optional[str] = None,
orig_tn: Optional[str] = None,
dest_tn: Optional[str] = None,
) -> str
Build a SHAKEN Identity header and add it to request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
The outbound SIP request. |
required |
attestation
|
str
|
|
'A'
|
origid
|
Optional[str]
|
UUID origin identifier; a fresh v4 is generated if |
None
|
orig_tn
|
Optional[str]
|
Originating TN; defaults to the From user part. |
None
|
dest_tn
|
Optional[str]
|
Destination TN; defaults to the To / R-URI user part. |
None
|
Returns:
| Type | Description |
|---|---|
str
|
The |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if signing is not configured. |
ValueError
|
if the orig/dest TN cannot be determined. |
sign_div
¶
sign_div(
request: Any,
orig_tn: Optional[str] = None,
dest_tn: Optional[str] = None,
div_tn: Optional[str] = None,
) -> None
Build a diverted-call (div) Identity header (RFC 8946).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
The outbound (retargeted) SIP request. |
required |
orig_tn
|
Optional[str]
|
Originating TN; defaults to the From user part. |
None
|
dest_tn
|
Optional[str]
|
New destination TN; defaults to the To / R-URI user part. |
None
|
div_tn
|
Optional[str]
|
Diverting TN; defaults to the History-Info / Diversion user. |
None
|
verify
¶
Verify the Identity header(s) on request.
Returns a :class:MockStirResult. By default returns a passing result
when an Identity header is present, else No-TN-Validation;
override with :meth:set_verify_result.
Raises:
| Type | Description |
|---|---|
RuntimeError
|
if verification is not configured. |
apply_verstat
¶
Stamp the verstat parameter onto the asserted identity
(P-Asserted-Identity if present, else From) per ATIS-1000074 §5.3.1.
set_verify_result
¶
set_verify_result(
verstat: str = "TN-Validation-Passed",
passed: bool = True,
attestation: Optional[str] = None,
origid: Optional[str] = None,
orig_tn: Optional[str] = None,
reason: str = "ok",
passports: Optional[list[dict[str, Any]]] = None,
) -> None
Pin the result returned by the next :meth:verify call(s).
StirResult¶
The outcome of stir.verify(...).
Result of :meth:MockStir.verify — mirrors the Rust StirResult.
Attributes:
| Name | Type | Description |
|---|---|---|
verstat |
|
|
passed |
|
|
attestation |
|
|
origid |
|
|
orig_tn |
originating TN from the SHAKEN PASSporT. |
|
reason |
human-readable diagnostic / failure cause. |
|
passports |
list[dict[str, Any]]
|
decoded PASSporT claim dicts. |
passports
property
¶
Decoded claim sets of every PASSporT that parsed.