Skip to content

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

auth:
  algorithms: ["MD5", "SHA-256", "SHA-512-256"]   # the default

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:

auth:
  algorithms: ["MD5"]

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 is 0 by default, and visible in siphon_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

generate_nonce() -> str

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

validate_nonce(nonce: str) -> bool

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_user(realm: str, username: str, password: str) -> None

Add credentials for testing (test helper).

Parameters:

Name Type Description Default
realm str

Auth realm (e.g. "example.com").

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 Request (@proxy.on_request) or B2BUA Call (@b2bua.on_invite).

required
realm Optional[str]

Auth realm (e.g. "example.com").

None
password Optional[str]

Verify against this plaintext secret instead of the configured backend. Mutually exclusive with ha1.

None
ha1 Optional[str]

Verify against this already-computed H(A1) instead of the configured backend. Mutually exclusive with password.

None

Returns:

Type Description
bool

True if authenticated, False if challenge was sent.

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 Request or B2BUA Call.

required
realm Optional[str]

Auth realm.

None
password Optional[str]

Verify against this plaintext secret instead of the configured backend. Mutually exclusive with ha1.

None
ha1 Optional[str]

Verify against this already-computed H(A1) instead of the configured backend. Mutually exclusive with password.

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

require_ims_digest(
    request: Any, realm: Optional[str] = None
) -> bool

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

True if credentials are valid, False if a 401 challenge was sent.

require_aka_digest

require_aka_digest(
    request: Any, realm: Optional[str] = None
) -> bool

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

True if credentials are valid, False if a 401 challenge was sent.

stamp_integrity_protected

stamp_integrity_protected(request: Any) -> Optional[str]

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

required

Returns:

Type Description
Optional[str]

"yes" or "no" as stamped on the first Authorization

Optional[str]

header, or None when the request has none.

verify_integrity_protected

verify_integrity_protected(request: Any) -> bool

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 Authorization header carrying integrity-protected says "yes", "tls-yes" or "ip-assoc-yes", and
  • that header's username is the identity that authenticated a live binding of the To AoR (:attr:Contact.auth_user, with implicit-set aliases resolved the way registrar.lookup does).

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

required

Returns:

Type Description
bool

True when the REGISTER may be accepted without a challenge.

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 Request or B2BUA Call.

required
realm Optional[str]

Auth realm. When None, the realm in the header is used.

None
password Optional[str]

Plaintext secret to verify against. Mutually exclusive with ha1.

None
ha1 Optional[str]

Already-computed H(A1) to verify against. Mutually exclusive with password.

None

Returns:

Type Description
bool

True if valid credentials are present.

Raises:

Type Description
ValueError

if both password and ha1 are given, or if the header names a digest algorithm this Python cannot compute.

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

set_pcscf_families(
    v4: bool = True, v6: bool = True
) -> None

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

activate(
    *, hard_lifetime_secs: Optional[int] = None
) -> None

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" / "B" / "C" (full / partial / gateway).

'A'
origid Optional[str]

UUID origin identifier; a fresh v4 is generated if None.

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 origid used.

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(request: Any) -> MockStirResult

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

apply_verstat(request: Any, result: MockStirResult) -> None

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

"TN-Validation-Passed" | "TN-Validation-Failed" | "No-TN-Validation" (ATIS-1000074 §5.3.1).

passed

True only when the SHAKEN PASSporT validated end to end.

attestation

"A" / "B" / "C" from the SHAKEN PASSporT.

origid

origid (UUID) from the SHAKEN PASSporT.

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

passports: list[dict[str, Any]]

Decoded claim sets of every PASSporT that parsed.