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. require_ims_digest also takes server_name=, the S-CSCF's own SIP URI, which the Multimedia-Auth-Request it sends to the HSS requires; see Cx: registration.

require_aka_digest computes the vector itself from auth.aka_credentials (Milenage), so it also keeps what an HSS would: a sequence number per subscriber. Each challenge carries a later SQN (the next SEQ, IND 0, TS 33.102 annex C), starting from the credential's optional sqn (12 hex characters, default 000000000001). A UE that finds the SQN out of range answers with auts=; siphon checks its MAC-S against the RAND of the challenge it answers, and on a match takes up the UE's SQN and sends a fresh 401, on a mismatch answers 403 (TS 33.102 6.3.5). The counters are in memory: after a restart the first challenge carries the configured sqn again, and a UE further along re-synchronises through that AUTS.

auth:
  aka_credentials:
    "001010000000001":
      k: "465b5ce8b199b49faa5f0a2ee238a6bc"
      op: "cdc202d5123e20f62b6d676ac72cb318"
      amf: "8000"
      sqn: "000000000020"   # optional: SQN of the first challenge

IMS AKA or SIP Digest¶

require_ims_digest authenticates a REGISTER against the HSS over Cx, with IMS-AKA or with SIP Digest (TS 33.203 Annex N). Which one is decided the way TS 24.229 clause 5.4.1.1 lists it, from the integrity-protected parameter the P-CSCF writes into the Authorization header and, for a REGISTER without that header, from P-Access-Network-Info:

The REGISTER	Item of clause 5.4.1.1	SIP-Authentication-Scheme asked of the HSS	Then
`integrity-protected="no"`	1	`Digest-AKAv1-MD5`	`AKAv1-MD5` challenge
`integrity-protected="yes"`	2	`Digest-AKAv1-MD5` when a vector is needed	the response is checked against the stored vector, a new challenge otherwise
`integrity-protected="tls-connected"`	2A	none	`403`: IMS AKA over TLS (AKAv2) is not implemented
no Authorization, P-Access-Network-Info with `network-provided` naming a non-3GPP access	5	`SIP Digest`	`Digest` challenge
no Authorization otherwise	none applies (4, 6, 7 and 8 need NASS-IMS bundled or GPRS-IMS-Bundled authentication)	`Unknown`	the scheme the HSS answers with
Authorization without `integrity-protected`	9	`Unknown`	the scheme the HSS answers with
`tls-pending`, `tls-yes`, `ip-assoc-pending`, `ip-assoc-yes`	10	`SIP Digest` when H(A1) is needed	the response is checked against the stored H(A1), a `Digest` challenge otherwise
`integrity-protected="auth-done"`	11	none	`403`: trusted node authentication is not implemented
any other value	none	none	`403`

A REGISTER whose nonce is one of a challenge this S-CSCF issued and still holds is handled under the mechanism of that challenge, whatever the parameter says. That is a deliberate difference from item 1, which has the S-CSCF ignore a response next to integrity-protected="no": a deployment without IPsec between UE and P-CSCF can still complete IMS-AKA.

To Unknown a conformant HSS answers SIP Digest or NASS-Bundled, and DIAMETER_ERROR_AUTH_SCHEME_NOT_SUPPORTED (5006) for an IMS-AKA subscriber (TS 29.228 clause 6.3.1 step 4). NASS-Bundled and Early-IMS-Security are refused with 403. An IMS-AKA item in answer to Unknown is challenged as IMS-AKA. So IMS-AKA needs a P-CSCF that writes integrity-protected, which auth.stamp_integrity_protected does.

SIP Digest¶

The HSS returns H(A1) in SIP-Digest-Authenticate, never the password:

AVP	Code, vendor	What `require_ims_digest` does with it
`Digest-Realm`	104, none	the `realm` of the challenge (not the `realm=` argument, H(A1) is bound to it)
`Digest-Algorithm`	111, none	optional, must be `MD5`
`Digest-QoP`	110, none	must be `auth`; the challenge carries `qop="auth"`
`Digest-HA1`	121, none	H(A1) for MD5, 32 hexadecimal digits
`Alternate-Digest-Algorithm`	662, 3GPP	optional, `SHA-256` or `SHA-512-256`
`Alternate-Digest-HA1`	663, 3GPP	H(A1) for that algorithm, 64 hexadecimal digits

The 401 carries one WWW-Authenticate: Digest per algorithm there is a hash for, the alternate one first, with one nonce generated by siphon (TS 24.229 clause 5.4.1.2.1B). The REGISTER that answers it is checked as clause 5.4.1.2.2A lists: same Call-ID as the challenged REGISTER, the username that was challenged, the realm and an algorithm of the challenge, qop=auth with cnonce and nc, and the response computed from H(A1) per RFC 7616.

The REGISTER carries	Answer
a response that verifies	`True`, `request.auth_user` set, `Authentication-Info: qop=auth, rspauth="...", cnonce="...", nc=...` on the 200
a response that does not verify, or misses one of the checks above	`403 Forbidden`, no challenge (TS 33.203 clause N.2.2.1), the nonce is dropped
a valid response with an `nc` not higher than the last one accepted	`401` with a fresh nonce and `stale=true`
a valid response under a nonce that expired or was never issued	`401` with a fresh nonce and `stale=true`, after a new request to the HSS
a response that is valid for neither	`403 Forbidden`

A nonce waits 120 seconds for its first answer. Once a response under it has verified it stays valid for nonce_lifetime seconds (default 3600), renewed by every REGISTER that verifies, so a re-registration with the next nc needs no challenge and no request to the HSS. No nextnonce is sent. A nonce belongs to one public identity: under another one the HSS is asked again.

The HSS side, as a @diameter.on_request handler builds it:

answer = request.answer(2001)
answer.set_avp("User-Name", private_identity)
answer.set_avp("Public-Identity", public_identity)
answer.set_avp("SIP-Number-Auth-Items", 1)
answer.insert_avp("SIP-Auth-Data-Item", [
    ("SIP-Authentication-Scheme", "SIP Digest"),
    ("SIP-Digest-Authenticate", [
        ("Digest-Realm", realm),
        ("Digest-Algorithm", "MD5"),
        ("Digest-QoP", "auth"),
        ("Digest-HA1", md5_hex(f"{private_identity}:{realm}:{password}")),
        ("Alternate-Digest-Algorithm", "SHA-256"),
        ("Alternate-Digest-HA1", sha256_hex(f"{private_identity}:{realm}:{password}")),
    ]),
])

A P-CSCF in front of a SIP Digest client removes the integrity-protected the client sent and writes its own with auth.set_integrity_protected(request, value): nothing (None) on a REGISTER without a challenge response, "ip-assoc-pending" on one with a response and no IP association yet, "ip-assoc-yes" once there is one, and "tls-pending" / "tls-yes" over TLS (TS 24.229 clauses 5.2.2.3 and 5.2.2.4).

What the HSS has to carry in the request, and what each of its answers leads to, is under Cx: registration.

`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,
    server_name: Optional[str] = None,
    nonce_lifetime: int = 3600,
) -> bool

IMS authentication via Diameter Cx MAR/MAA: IMS AKA or SIP Digest.

Sends a Multimedia-Auth-Request to the HSS the Cx route names and uses what it returns to challenge or verify the UE. The mock has no HSS, so it checks the request against the test's users.

In siphon the mechanism is the one TS 24.229 clause 5.4.1.1 selects from the REGISTER:

integrity-protected yes or no: asks the HSS for Digest-AKAv1-MD5 and challenges with AKAv1-MD5;
tls-pending, tls-yes, ip-assoc-pending or ip-assoc-yes: asks for SIP Digest and challenges with Digest;
an Authorization header without the parameter: asks for Unknown and challenges in the scheme the HSS answers with;
no Authorization header and a P-Access-Network-Info with network-provided naming a non-3GPP access: SIP Digest;
no Authorization header otherwise: Unknown;
tls-connected or auth-done: refused with 403, the HSS is not asked.

A REGISTER that answers a challenge siphon issued is handled under the mechanism of that challenge. So IMS AKA needs a P-CSCF that calls :meth:stamp_integrity_protected on every REGISTER.

For SIP Digest (TS 33.203 Annex N) the HSS returns H(A1) and the 401 carries one Digest challenge per algorithm it sent a hash for, SHA-256 before MD5, with a nonce siphon generates, the realm of the HSS and qop="auth". A response that verifies sets Authentication-Info (qop, rspauth, cnonce, nc) on the 200. The nonce then stays valid for nonce_lifetime seconds as long as each REGISTER raises nc. A response that does not verify is refused with 403. One under an expired nonce, or with an nc that is not higher than the last, is challenged again with stale=true.

The request names the public user identity of the To header (TS 24.229 clause 5.4.1.2.1). A successful answer in another scheme than the one named, or without the vector or the H(A1) of its scheme, is not turned into a challenge.

Parameters:

Name	Type	Description	Default
`request`	`Any`	The REGISTER.	required
`realm`	`Optional[str]`	Digest realm of an IMS AKA challenge. A SIP Digest challenge carries the realm of the HSS.	`None`
`server_name`	`Optional[str]`	This S-CSCF's SIP URI, sent as Server-Name. The MAR requires it (TS 29.229 clause 6.1.7), so pass the URI given to `diameter.cx_sar`. The mock accepts and ignores it.	`None`
`nonce_lifetime`	`int`	Seconds a SIP Digest nonce stays valid after a response under it verified. The mock accepts and ignores it.	`3600`

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
`bool`	sent or a 403 was set.

Raises:

Type	Description
`RuntimeError`	in siphon, when no Diameter connection is available, the request got no answer, or the answer carries another authentication scheme than the one asked for or nothing to challenge with.

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.

Each subscriber has its own SQN, starting at the credential's optional sqn (12 hex chars, default 000000000001) and advancing with every challenge (TS 33.102 annex C). An Authorization carrying auts= (the UE found the SQN out of range) is checked against the RAND of the challenge it answers: a valid MAC-S re-bases the SQN and a fresh 401 goes out, an invalid one is answered 403 (TS 33.102 6.3.5). The mock does not model SQN or AUTS.

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 or 403 was set.

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.

set_integrity_protected ¶

set_integrity_protected(
    request: Any, value: Optional[str]
) -> bool

P-CSCF: set or remove integrity-protected in every Authorization header.

For the mechanisms :meth:stamp_integrity_protected does not decide. With SIP Digest the P-CSCF writes (3GPP TS 24.229 clauses 5.2.2.3 and 5.2.2.4):

nothing on a REGISTER without a challenge response that maps to no IP association or TLS session: pass None, which also removes what the UE sent (clause 5.2.2.1 step 4B);
"ip-assoc-pending" on one with a challenge response and no IP association yet, "ip-assoc-yes" on one that maps to an existing IP association;
"tls-pending" and "tls-yes" for the same two cases over TLS.

Example::

authorization = request.get_header("Authorization") or ""
answered = 'response="' in authorization and 'response=""' not in authorization
auth.set_integrity_protected(
    request, "ip-assoc-pending" if answered else None)
request.relay()

Parameters:

Name	Type	Description	Default
`request`	`Any`	The REGISTER `Request`.	required
`value`	`Optional[str]`	One of `"yes"`, `"no"`, `"tls-pending"`, `"tls-yes"`, `"ip-assoc-pending"`, `"ip-assoc-yes"`, `"auth-done"`, `"tls-connected"`, or `None` to remove the parameter.	required

Returns:

Type	Description
`bool`	`False`, changing nothing, when the request has no
`bool`	`Authorization` header; `True` otherwise.

Raises:

Type	Description
`ValueError`	for a value table 7.2A.2 does not list.

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.

pcscf_ports_c `property` `writable` ¶

pcscf_ports_c: list[int]

Every configured P-CSCF protected client port, in order.

ipsec.pcscf_port_c in siphon.yaml takes one port or a list. Each new SA pair takes the first port no other live pair of the same UE holds, so an authenticated re-registration changes port_pc (3GPP TS 33.203 §7.4.0); another UE may hold the same port. Advertise the pair's own as port-c::

p = pending.security_server_params()
reply.set_header("Security-Server", f"...; port-c={p.port_c}; port-s={p.port_s}")

In the mock, assign a list to model a pool; pcscf_port_c follows its first entry.

delete `async` ¶

delete(
    ue_addr: str,
    ue_port_c: Optional[int] = None,
    *,
    delay_secs: float = 0.0
) -> int

Delete the security associations of the UE at ue_addr.

The pair whose UE client port is ue_port_c, or every pair of the UE when it is None, goes with the stream connections between the protected ports and the UE's. Returns how many pairs were deleted, or with delay_secs how many were scheduled.

siphon already does this itself (3GPP TS 33.203 clause 7.4.2a, TS 24.229 clauses 5.2.5.1 and 5.2.5.2): once the server transaction of a 2xx to a de-REGISTER that removed the UE's last binding is over, and on the sweep after the registrar loses the last binding that was registered over a pair (a script removing it on a reg-event NOTIFY, or its expiry), 32 s on. Call this for what siphon cannot see. delay_secs keeps the pair that long first, so the NOTIFY of a network-initiated de-registration still reaches the UE.

Example::

await ipsec.delete(request.source_ip, request.source_port,
                   delay_secs=5)

In the mock, matching pairs from allocate() are marked cleaned (immediately, whatever the delay) and the call is recorded in :attr:deleted.

Raises:

Type	Description
`ValueError`	`ue_addr` is not an IP literal, or `delay_secs` is negative or not finite.

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.

When security associations are deleted¶

A pair is taken into use by PendingSA.activate() on the 200 to the REGISTER that authenticated it. siphon deletes it, with the stream connections between the protected ports and the UE's (its own are reset), when:

What happened	When the pair goes
A 2xx to a de-REGISTER (`Expires: 0`, or every Contact at `expires=0`) is sent to the UE, relayed or answered by the script, and no live registrar binding was registered over the pair any more	When the server transaction is over: Timer J (64*T1) over UDP, T1 over TCP (TS 24.229 clause 5.2.5.1)
The registrar loses the last binding registered over a pair without a de-REGISTER through siphon: the script removed it on a reg-event NOTIFY, or it expired	64*T1 after the next sweep sees it, so the NOTIFY to the UE still gets through (TS 24.229 clause 5.2.5.2)
A re-registration's new pair is activated	The old pair the REGISTER came over after 32 s, for the transactions pending on it; any other old pair at once (TS 33.203 clause 7.4.2a)
The script calls `await ipsec.delete(ue_addr, ue_port_c, delay_secs=...)`	After `delay_secs`, or at once

A pair goes only when no other identity of the UE is still registered over it.

With a single ipsec.pcscf_port_c, the old and new pair of a re-registration share the policies of the UE's protected server port; deleting one leaves the other's in place, and the shared outbound policy moves to the new pair as the 200 leaves. With a list (pcscf_port_c: [5064, 5068]), each new pair takes a port no other pair of the UE holds (TS 33.203 clause 7.4.0), so nothing is shared: the 401 goes over the old pair, the 200 over the new, requests to the UE move to the new pair once a message arrives over it, and the old pair then goes after 32 s (clause 7.4.2a). If the old pair goes first, the new one is taken into use.

After a restart¶

Security associations live and die with the P-CSCF process that negotiated them (TS 33.203). siphon never persists IK/CK, SPIs or any other SA material, so a restart can't reinstall them, even though the registrar bindings come back from a Redis or PostgreSQL backend. The UE still holds its side of the SAs.

A restored binding that was protected (its REGISTER arrived on any ipsec.pcscf_port_c or ipsec.pcscf_port_s) and has no SA pair in the new process is therefore not restored as live. At startup, before any traffic, siphon:

drops it from the registrar and removes it from the backend through the normal de-registration path, so the next boot doesn't restore it again (a binding another instance accepted on a shared backend leaves only this node's memory; its row stays with the instance that holds the SAs);
logs one warning per binding (AoR, UE address and ports, no key material) and counts it in siphon_registrar_restored_without_sa_total;
bars the UE's protected client and server ports from requests in clear until the binding would have expired (plus 32 s). Any request that still targets one of them while no SA pair covers it is refused at egress, logged at error and counted in siphon_ipsec_unprotected_sends_refused_total. Responses are not barred, and a new SA pair on those ports lifts the bar.

With the binding gone, a terminating request finds no contact and the liveness sweep has nothing to probe. Recovery is the specified one: the UE's next protected request goes unanswered, and it performs an initial registration (TS 24.229 clause 5.1.1.4.1) with fresh AKA and new SAs. To get UEs back sooner, the network can run P-CSCF restoration (TS 23.380 clause 5); siphon doesn't implement it.

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

On a re-authentication siphon keeps sending over the UE's set in use until the 200 to the REGISTER that came over this pair leaves: that 200, and what follows, go over this pair (TS 24.229 clause 5.2.2.2). The dispatcher does the switch, whether or not this is called.

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.

When a newer challenge replaced this pair on the UE's protected client port, siphon does nothing in the kernel: the newer pair is not re-pinned or superseded through this one.

In the mock, this only updates self.expires_secs so tests can assert the script wired the grant through correctly.

cleanup `async` ¶

cleanup() -> None

Tear down this pair's four XFRM states and policies.

Only this pair, named by its own SPIs: when a retransmitted REGISTER was challenged again, the newer pair that replaced this one on the UE's protected client port is left alone (3GPP TS 33.203 clause 7.4). Idempotent. ipsec.stash runs it on a stashed entry that expires or is displaced.

refresh `async` ¶

refresh(av_new: MockAuthVectorHandle) -> None

Re-key the pair with a new authentication vector.

Raises ValueError when the PendingSA was cleaned up, or when a newer challenge replaced its pair on the UE's protected client port (siphon only; the mock does not model replacement).

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

Auth & security¶

Challenging in B2BUA mode¶

When the credential is not the subscriber identity¶

IMS AKA or SIP Digest¶

SIP Digest¶

auth namespace¶

Which algorithms a challenge offers¶

Issuing your own challenge¶

Verifying against a credential you hold¶

generate_nonce ¶

validate_nonce ¶

add_user ¶

require_www_digest ¶

require_proxy_digest ¶

require_digest ¶

require_ims_digest ¶

require_aka_digest ¶

stamp_integrity_protected ¶

set_integrity_protected ¶

verify_integrity_protected ¶

verify_digest ¶

ipsec namespace¶

pcscf_ports_c property writable ¶

delete async ¶

set_allocate_failure ¶

set_pcscf_families ¶

When security associations are deleted¶

After a restart¶

SecurityOffer¶

Transform¶

AuthVectorHandle¶

PendingSA¶

activate ¶

cleanup async ¶

refresh async ¶

SecurityServerParams¶

SAHandle¶

stir namespace¶

sign ¶

sign_div ¶

verify ¶

apply_verstat ¶

set_verify_result ¶

StirResult¶

passports property ¶

`auth` namespace¶

`ipsec` namespace¶

pcscf_ports_c `property` `writable` ¶

delete `async` ¶

`SecurityOffer`¶

`Transform`¶

`AuthVectorHandle`¶

`PendingSA`¶

cleanup `async` ¶

refresh `async` ¶

`SecurityServerParams`¶

`SAHandle`¶

`stir` namespace¶

`StirResult`¶

passports `property` ¶