Registrar¶

The registrar namespace is the location service: it saves contact bindings, looks them up, and handles the IMS implicit registration set, service routes, and pending/confirm flows. The registration namespace is the opposite direction — outbound REGISTER to upstream carriers and SBCs.

from siphon import registrar

@proxy.on_request("REGISTER")
def register(request):
    if auth.verify_digest(request, "example.com"):
        registrar.save(request)   # saves contacts and sends 200 OK
    else:
        auth.require_www_digest(request, "example.com")

`registrar` namespace¶

Mock registrar with an in-memory contact store.

Pre-populate contacts for testing::

from siphon import registrar
registrar.add_contact("sip:alice@example.com",
                      Contact(uri="sip:alice@192.168.1.5:5060"))

Then your script's registrar.lookup() will find them.

save ¶

save(
    request: Any,
    force: bool = False,
    aliases: Optional[list[str]] = None,
    flow_token: Optional[str] = None,
) -> bool

Save contact bindings from a REGISTER request and send the 200 OK reply.

Stores the contact bindings and automatically sends a 200 OK reply to the REGISTER request with the granted Expires header — the script must not call request.reply(200, "OK") afterwards.

In the mock, extracts the To URI as AoR and stores a default contact binding.

Parameters:

Name	Type	Description	Default
`request`	`Any`	The REGISTER request object.	required
`force`	`bool`	If `True`, evict all existing contacts first.	`False`
`aliases`	`Optional[list[str]]`	IMS implicit registration set (3GPP TS 23.228) — every URI in the list becomes an alias of this AoR, so subsequent `registrar.lookup(alias)` calls resolve to the same contacts. Empty / `None` is a no-op; clear an existing set with `registrar.set_associated_uris(aor, [])`.	`None`
`flow_token`	`Optional[str]`	Opaque proxy-side token to attach to every contact saved by this call. Captures the inbound flow so subsequent `registrar.lookup_by_token(flow_token)` resolves back to this binding and the script can `request.relay(flow=binding.flow)` for P-CSCF MT routing (RFC 3327 §5 / TS 24.229 §5.2.7.2).	`None`

Returns:

Type	Description
`bool`	`True` on success.

Example::

if request.method == "REGISTER":
    if not auth.require_digest(request, realm=DOMAIN):
        return
    # Generate an opaque token, write it into Path so MT
    # requests come back with it on the topmost Route.
    token = secrets.token_urlsafe(16)
    request.add_pcscf_path(token)
    registrar.save(request, flow_token=token)
    return

save_proxy ¶

save_proxy(
    request: Any,
    reply: Any,
    aliases: Optional[list[str]] = None,
    flow_token: Optional[str] = None,
) -> bool

Cache a binding on a proxy after the upstream registrar accepted it.

Use on a proxy (e.g. P-CSCF in IMS) that wants a local copy of a UE's binding for routing terminating requests, where the actual REGISTER was forwarded to a registrar of record (e.g. S-CSCF) and a 200 OK has just come back.

Differs from :meth:save in three ways:

The contact lifetime is read from the reply's Expires header (the registrar's grant per RFC 3261 §10.3 step 8), not the request's (the UE's ask). UEs commonly ask for 600000 s; the registrar caps to a sensible value, and mirroring that cap locally is incorrect — the proxy must trust the upstream's decision.
The local max_expires cap is not applied. The registrar of record has already capped, and a tighter local cap would expire the proxy cache before the upstream binding, opening a window where MT requests would 404 against an entry the registrar still considers live.
No 200 OK is generated — the proxy will relay the upstream's response itself.

A grace of ~32 s (RFC 3261 Timer F = 64·T1) is added on top so a NOTIFY[reg-event;state=terminated] from the registrar at expiry has a transaction-timer window to land before the proxy forgets.

Expires: 0 on the reply clears the binding (de-REGISTER path).

Parameters:

Name	Type	Description	Default
`request`	`Any`	The original REGISTER (read for AoR + Contact list).	required
`reply`	`Any`	The upstream 200 OK (read for granted `Expires`).	required
`aliases`	`Optional[list[str]]`	IMS implicit registration set, same shape as :meth:`save` `aliases=` — see that method's docs.	`None`

Raises:

Type	Description
`ValueError`	when the reply has no parseable `Expires` header (the registrar of record must include the granted `Expires` per RFC 3261 §10.3 step 8).

Example::

@proxy.on_reply
def on_reply(request, reply):
    if request.method == "REGISTER" and reply.status_code == 200:
        registrar.save_proxy(request, reply,
                             aliases=raw_uris or [])
    reply.relay()

lookup ¶

lookup(uri: Union[str, SipUri]) -> list[Contact]

Look up routable contacts for an address-of-record.

Returns only UE-side bindings (kind == "ue"). AS-side capability records — captured via :meth:save_as_contact — are excluded so a misrouted MT INVITE never goes to an AS (TS 24.229 §5.4.2.1.2). See :func:registrar.reginfo_xml for the merged view that surfaces AS feature tags.

If the URI is an alias of an IMS implicit registration set, resolves to the primary's contacts (matching production registrar.lookup behaviour).

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR as string or :class:`SipUri`.	required

Returns:

Type	Description
`list[Contact]`	List of UE-side :class:`Contact` objects sorted by q-value
`list[Contact]`	(descending). Empty list if no UE contacts registered.

lookup_contact ¶

lookup_contact(uri: Union[str, SipUri]) -> list[Contact]

Reverse lookup by Contact URI.

:meth:lookup resolves a logical address (user@domain); this resolves a physical one — it returns every registered UE-side binding whose stored Contact matches uri (user + host + port; URI parameters and default ports are ignored).

Use it on the terminating edge when the only thing you have is the contact. A common case: a PBX in front of siphon retargets the INVITE straight at the cached Contact and loose-routes it back, so call.ruri is the contact (sip:1001@203.0.113.7:17514), not the registration AoR (sip:1001@pbx.example). lookup() keys on the AoR and misses; lookup_contact() matches the binding regardless of the AoR domain::

@b2bua.on_invite
def route(call):
    if not registrar.lookup_contact(str(call.ruri)):
        call.reject(404, "No extension Found")
        return
    call.dial(str(call.ruri))

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	Contact URI as string or :class:`SipUri`.	required

Returns:

Type	Description
`list[Contact]`	List of matching UE-side :class:`Contact` objects sorted by
`list[Contact]`	q-value (descending). AS-side capability records are
`list[Contact]`	excluded, matching :meth:`lookup`. Empty list if no binding
`list[Contact]`	has that contact.

save_as_contact ¶

save_as_contact(
    aor: Union[str, SipUri],
    reply: Any,
    expires_secs: Optional[int] = None,
) -> bool

Save AS-side capability contacts from a 3PR 200 OK (3GPP TS 24.229 §5.4.2.1.2).

The S-CSCF runs iFC, fires a third-party REGISTER at each matched AS, and receives a 200 OK whose Contact: header carries the AS's URI plus RFC 3840 feature tags (+g.3gpp.smsip, +g.3gpp.icsi-ref, …). Calling this from @proxy.on_reply (or after a proxy.send_request(..., wait_for_response=True)) caches every such Contact alongside the UE's own bindings so the next reg-event NOTIFY surfaces them to watchers.

AS contacts are stored with kind="as" and excluded from :meth:lookup — they only exist to be advertised in reg-event NOTIFY bodies (no MT INVITE ever routes to them).

Parameters:

Name	Type	Description	Default
`aor`	`Union[str, SipUri]`	IMPU the AS responded for.	required
`reply`	`Any`	200 OK from the AS. Its `Contact:` headers are walked; `+sip.instance` / `reg-id` are NOT broken out (no GRUU semantic on the AS side).	required
`expires_secs`	`Optional[int]`	lifetime for the cached AS contact. When `None`, falls back to the reply's `Expires` header (raises `ValueError` if absent).	`None`

Returns:

Type	Description
`bool`	`True` if at least one Contact was stored; `False` if
`bool`	the reply had no Contact headers, or the AoR has no UE-side
`bool`	binding (the registrar refuses to store an AS capability
`bool`	record against an unregistered user).

Example::

@proxy.on_reply
def on_reply(request, reply):
    if request.method == "REGISTER" and reply.status_code == 200:
        registrar.save_as_contact(str(request.to_uri), reply)
    reply.relay()

lookup_by_token ¶

lookup_by_token(token: str) -> Optional[Contact]

Resolve an opaque flow-token previously attached via registrar.save(flow_token=...) to its bound contact.

Returns None when the token is unknown, the binding has expired, or no contact in the resolved AoR carries this token.

Used by P-CSCF MT routing (RFC 3327 §5 / TS 24.229 §5.2.7.2): the proxy advertised a Path URI of the form <sip:TOKEN@pcscf;lr>; on the MT request, after loose_route() consumed that Route, request.consumed_route_user exposes the token and this method resolves it back to the binding so the script can call request.relay(flow=binding.flow).

Parameters:

Name	Type	Description	Default
`token`	`str`	Opaque token previously passed to `registrar.save(flow_token=...)`.	required

Returns:

Type	Description
`Optional[Contact]`	The matching :class:`Contact` (with `.flow` populated)
`Optional[Contact]`	or `None`.

is_registered ¶

is_registered(uri: Union[str, SipUri]) -> bool

Check if a URI has any registered UE-side contacts.

Mirrors the Rust-side semantic — AS capability records don't register a user.

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR as string or :class:`SipUri`.	required

is_registered_contact ¶

is_registered_contact(uri: Union[str, SipUri]) -> bool

Whether any registered binding has a Contact URI matching uri.

Contact-keyed twin of :meth:is_registered; see :meth:lookup_contact for when the terminating edge needs to match on the contact rather than the AoR.

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	Contact URI as string or :class:`SipUri`.	required

aor_count `async` ¶

aor_count() -> int

Number of currently registered AoRs across the deployment.

Async — when a persistent backend (Redis, Postgres) is configured the Rust implementation queries the backend so the count is authoritative across all siphon instances sharing it. Without a backend it returns the local in-memory count.

The mock simply counts the in-memory store.

Returns:

Type	Description
`int`	Number of distinct AoRs that currently have at least one
`int`	non-expired contact binding.

Example::

from siphon import registrar, metrics, timer

gauge = metrics.gauge("siphon_aors_registered",
                      "Currently registered AoRs")

@timer.every(seconds=15)
async def publish_aor_count():
    gauge.set(await registrar.aor_count())

expire ¶

expire(uri: Union[str, SipUri]) -> None

Force-expire all contacts for a URI.

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR to expire.	required

remove ¶

remove(uri: Union[str, SipUri]) -> None

Remove all contacts for a URI (deregistration).

Alias for :meth:expire -- used from RTR handlers.

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR to remove.	required

save_pending ¶

save_pending(request: Any) -> None

Save contacts in pending state (IMS: awaiting SAR confirmation).

Parameters:

Name	Type	Description	Default
`request`	`Any`	The REGISTER request to extract contacts from.	required

confirm_pending ¶

confirm_pending(uri: Union[str, SipUri]) -> None

Confirm pending contacts (IMS: SAR succeeded).

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR to confirm.	required

asserted_identity ¶

asserted_identity(uri: Union[str, SipUri]) -> Optional[str]

Look up stored P-Asserted-Identity for a URI.

Returns:

Type	Description
`Optional[str]`	Identity string if stored, otherwise `None`.

set_asserted_identity ¶

set_asserted_identity(aor: str, identity: str) -> None

Store P-Asserted-Identity for an AoR (test helper).

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address-of-record.	required
`identity`	`str`	P-Asserted-Identity value.	required

set_service_routes ¶

set_service_routes(aor: str, routes: list[str]) -> None

Store Service-Route headers for an AoR (RFC 3608).

Called after SAR success in the S-CSCF to record the routes that subsequent requests from this UE should traverse.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address-of-record string.	required
`routes`	`list[str]`	List of Route URI strings.	required

service_route ¶

service_route(uri: Union[str, SipUri]) -> list[str]

Get stored Service-Route headers for a URI (RFC 3608).

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR as string or :class:`SipUri`.	required

Returns:

Type	Description
`list[str]`	List of Route URI strings, or empty list.

set_associated_uris ¶

set_associated_uris(aor: str, uris: list[str]) -> None

Store P-Associated-URI list for an AoR and rebuild the derived alias index.

Each URI in uris becomes an alias of aor, so subsequent registrar.lookup(alias) / registrar.is_registered(alias) calls resolve to aor's contacts. Empty list clears both the AU list and every alias entry pointing at this primary.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address-of-record string (or any alias of it — the call is resolved to the primary).	required
`uris`	`list[str]`	List of P-Associated-URI strings.	required

associated_uris ¶

associated_uris(uri: Union[str, SipUri]) -> list[str]

Get stored P-Associated-URI list for a URI.

Parameters:

Name	Type	Description	Default
`uri`	`Union[str, SipUri]`	AoR as string or :class:`SipUri`.	required

Returns:

Type	Description
`list[str]`	List of P-Associated-URI strings, or empty list.

on_change `staticmethod` ¶

on_change(fn: Callable) -> Callable

Register a handler for registration state changes.

The handler receives (aor, event_type, contacts) where: - aor: str — Address of Record - event_type: str — "registered", "refreshed", "deregistered", or "expired" - contacts: list[Contact] — current contact bindings

Usage::

@registrar.on_change
def on_reg_change(aor, event_type, contacts):
    ...

reginfo_xml ¶

reginfo_xml(
    aor: str, state: str = "full", version: int = 0
) -> str

Generate RFC 3680 reginfo XML for an AoR.

Returns the XML document as a string. Includes both UE-side bindings and AS-side capability records (TS 24.229 §5.4.2.1.2) — the latter surface their RFC 3840 feature tags as <unknown-param> children (RFC 3680 §5.3.2).

Registration state is "active" when at least one UE-side contact exists, otherwise "terminated" (AS-only AoRs don't register a user).

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address of Record (e.g. `"sip:alice@example.com"`).	required
`state`	`str`	`"full"` or `"partial"` (default `"full"`).	`'full'`
`version`	`int`	reginfo version counter (default 0).	`0`

Returns:

Type	Description
`str`	XML string conforming to RFC 3680.

add_contact ¶

add_contact(aor: str, contact: Contact) -> None

Add a contact binding directly (test helper).

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address-of-record string (e.g. `"sip:alice@example.com"`).	required
`contact`	`Contact`	:class:`Contact` object to register.	required

clear ¶

clear() -> None

Remove all registrations (test helper).

`registration` namespace¶

Outbound REGISTER client for carrier / trunk registration.

Mock outbound registration namespace.

Manages outbound REGISTER bindings to upstream carriers/SBCs.

Example::

from siphon import registration

registration.add("sip:bob@carrier.com", "sip:registrar.carrier.com",
                  user="bob", password="pass123", interval=3600)
registration.remove("sip:bob@carrier.com")

for reg in registration.list():
    log.info(f"{reg['aor']}: {reg['state']}")

add ¶

add(
    aor: str,
    registrar: str,
    *,
    user: str,
    password: str = "",
    interval: Optional[int] = None,
    realm: Optional[str] = None,
    contact: Optional[str] = None,
    transport: Optional[str] = None,
    auth: Optional[str] = None,
    k: Optional[str] = None,
    op: Optional[str] = None,
    opc: Optional[str] = None,
    amf: Optional[str] = None,
    sqn: Optional[str] = None,
    ipsec: bool = False,
    ue_port_c: Optional[int] = None,
    ue_port_s: Optional[int] = None,
    ipsec_alg: Optional[str] = None,
    ipsec_ealg: Optional[str] = None,
    imei: Optional[str] = None,
    ims_features: Optional[list[str]] = None
) -> None

Add a new outbound registration.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address-of-Record (e.g. "sip:alice@carrier.com"). For IMS AKA this is the IMPU (e.g. "sip:001010000000001@ims.mnc01.mcc001.3gppnetwork.org").	required
`registrar`	`str`	Registrar URI (e.g. "sip:registrar.carrier.com:5060"). For IMS this is the P-CSCF.	required
`user`	`str`	Authentication username. For IMS AKA this is the IMPI.	required
`password`	`str`	Authentication password (digest only; unused for AKA).	`''`
`interval`	`Optional[int]`	Registration interval in seconds.	`None`
`realm`	`Optional[str]`	Optional realm hint (the home domain for IMS).	`None`
`contact`	`Optional[str]`	Optional Contact URI.	`None`
`transport`	`Optional[str]`	Transport protocol: "udp" (default), "tcp", "tls".	`None`
`auth`	`Optional[str]`	"digest" (default) or "aka" for IMS AKAv1-MD5 (RFC 3310 / 3GPP TS 33.203).	`None`
`k`	`Optional[str]`	Subscriber key K as 32 hex chars (required when auth="aka").	`None`
`op`	`Optional[str]`	Operator variant OP as 32 hex chars (supply op OR opc for AKA).	`None`
`opc`	`Optional[str]`	Pre-computed OPc as 32 hex chars (supply op OR opc for AKA).	`None`
`amf`	`Optional[str]`	Authentication Management Field as 4 hex chars (default "8000").	`None`
`sqn`	`Optional[str]`	Initial stored sequence number SQN_MS as 12 hex chars (default all-zeros — correct for a fresh soft-UE).	`None`
`ipsec`	`bool`	True to establish IPsec sec-agree with the P-CSCF (3GPP TS 33.203). Requires auth="aka", ue_port_c, ue_port_s.	`False`
`ue_port_c`	`Optional[int]`	UE protected client port (must also be a listen.udp port).	`None`
`ue_port_s`	`Optional[int]`	UE protected server port (must also be a listen.udp port).	`None`
`ipsec_alg`	`Optional[str]`	Offered integrity algorithm — "hmac-sha-1-96" (default), "hmac-md5-96", or "hmac-sha-256-128".	`None`
`ipsec_ealg`	`Optional[str]`	Offered encryption algorithm — "null" (default) or "aes-cbc".	`None`

Raises:

Type	Description
`ValueError`	when auth="aka" but `k` or an operator key (`op`/`opc`) is missing; or when ipsec=True without auth="aka" / ue_port_c / ue_port_s — mirroring the Rust binding.

remove ¶

remove(aor: str) -> bool

Remove an outbound registration by AoR.

refresh ¶

refresh(aor: str) -> bool

Force an immediate re-registration for an AoR.

list ¶

list() -> list[dict]

List all registrations with their current state.

Returns:

Type	Description
`list[dict]`	List of dicts with keys: aor, state, expires_in.

status ¶

status(aor: str) -> Optional[str]

Get the state of a specific registration.

count ¶

count() -> int

Number of configured registrations.

service_route ¶

service_route(aor: str) -> list[str]

The captured Service-Route set (RFC 3608) for an AoR — the Route a B2BUA prepends to MO calls so they traverse the originating S-CSCF. Empty in the mock unless populated on the entry dict by a test.

associated_uris ¶

associated_uris(aor: str) -> list[str]

The P-Associated-URI list (implicit registration set) for an AoR.

flow ¶

flow(aor: str, ue_ip: str)

A :class:Flow over the UE→P-CSCF IPsec SA for MO call.dial.

Real runtime returns None until the sec-agree handshake completes; the mock returns a Flow whenever the entry was added with ipsec=True (so MO handlers can be unit-tested), else None.

on_change `staticmethod` ¶

on_change(fn: Callable) -> Callable

Register a handler for outbound registration state changes.

The handler receives (aor, event_type, state) where: - aor: str -- Address of Record (e.g. "sip:trunk@carrier.com") - event_type: str -- "registered", "refreshed", "failed", or "deregistered" - state: dict -- {"expires_in": int, "failure_count": int, "registrar": str, "status_code": int} (status_code only present when event_type is "failed")

Usage::

@registration.on_change
def on_trunk_change(aor, event_type, state):
    ...

clear ¶

clear() -> None

Reset all registrations (test helper).

Registrar¶

registrar namespace¶

save ¶

save_proxy ¶

lookup ¶

lookup_contact ¶

save_as_contact ¶

lookup_by_token ¶

is_registered ¶

is_registered_contact ¶

aor_count async ¶

expire ¶

remove ¶

save_pending ¶

confirm_pending ¶

asserted_identity ¶

set_asserted_identity ¶

set_service_routes ¶

service_route ¶

set_associated_uris ¶

associated_uris ¶

on_change staticmethod ¶

reginfo_xml ¶

add_contact ¶

clear ¶

registration namespace¶

add ¶

remove ¶

refresh ¶

list ¶

status ¶

count ¶

service_route ¶

associated_uris ¶

flow ¶

on_change staticmethod ¶

clear ¶

`registrar` namespace¶

aor_count `async` ¶

on_change `staticmethod` ¶

`registration` namespace¶

on_change `staticmethod` ¶