Skip to content

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")
async def register(request):
    if await auth.verify_digest(request, "example.com"):
        registrar.save(request)   # saves contacts and sends 200 OK
    else:
        await auth.require_www_digest(request, "example.com")

When the registrar refuses a binding, registrar.save() answers the REGISTER itself, returns False and stores nothing. None of the REGISTER's Contacts are kept, and with force=True the existing bindings stay.

Refusal Answer
Expires below registrar.min_expires 423 Interval Too Brief with Min-Expires (RFC 3261 §10.3 step 7)
A new binding past registrar.max_contacts 503 Service Unavailable with Retry-After: seconds until the soonest held binding expires, 1 to max_expires
An AoR that is not a safe storage key 404 Not Found (RFC 3261 §10.3 step 3)

Each refusal is logged at warn and counted in siphon_registrar_refusals_total{reason} (interval_too_brief, too_many_contacts, invalid_aor). registrar.save_proxy() never answers a request, so it still raises ValueError for the same conditions.

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.

configure

configure(
    *,
    default_expires: Optional[int] = None,
    max_expires: Optional[int] = None,
    min_expires: Optional[int] = None,
    max_contacts: Optional[int] = None
) -> None

Set the registrar limits :meth:save enforces (test helper).

Mirrors the registrar: block of siphon.yaml. A limit not passed keeps its current value. :func:reset (and :meth:clear) restore the engine's defaults for a YAML that leaves them out: default_expires=3600, max_expires=7200, min_expires=60, max_contacts=10.

Parameters:

Name Type Description Default
default_expires Optional[int]

Lifetime granted when the REGISTER asks for none.

None
max_expires Optional[int]

Longest lifetime granted; a longer ask is capped.

None
min_expires Optional[int]

Shortest lifetime accepted; a shorter ask is answered 423 Interval Too Brief.

None
max_contacts Optional[int]

Bindings one AoR may hold; a new binding past it is answered 503 Service Unavailable.

None

Example::

harness.registrar.configure(max_contacts=1)
result = harness.send_request(
    "REGISTER", "sip:example.com", from_uri="sip:alice@example.com",
    headers={"Contact": "<sip:alice@192.0.2.11:5060>"},
)
assert result.status_code == 503

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 answer it.

Stores the REGISTER's Contacts under the AoR from its To header and sends the 200 OK. The script must not call request.reply(200, "OK") afterwards. Each Contact replaces the binding with the same +sip.instance (RFC 5627), else the one with the same URI, and expires=0 removes it. Its lifetime is the Contact's expires parameter, else the Expires header, else default_expires, capped at max_expires. Contact: * removes every binding. Bindings are stamped with :attr:~siphon_sdk.types.Contact.received from the request's source address, as the engine does, so contact.received or contact.uri resolves the same way it will there.

When the registrar refuses the binding, save() answers the REGISTER itself and returns False, so the script just returns:

  • Expires below min_expires: 423 Interval Too Brief with a Min-Expires header (RFC 3261 §10.3 step 7).
  • A new binding past max_contacts: 503 Service Unavailable with Retry-After set to the seconds until the soonest held binding expires (at least 1, at most max_expires, and max_expires when none is held). 503 rather than 403: a UA takes a 403 to its REGISTER as a bad credential and would only re-authenticate.
  • An AoR that is not a safe storage key: 404 Not Found.

A refused REGISTER stores nothing: none of its Contacts, and with force=True the existing bindings stay. No @registrar.on_change handler fires for it. Set the limits with :meth:configure.

Mock only: a REGISTER with no Contact header at all binds a synthetic sip:<ruri user>@<source ip>:5060 contact, so fixtures that never modelled a Contact keep registering something. On the engine such a REGISTER is a query and stores nothing.

Parameters:

Name Type Description Default
request Any

The REGISTER request object.

required
force bool

If True, replace all existing contacts, once the REGISTER has been accepted.

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 when the REGISTER was accepted and 200 OK sent;

bool

False when the registrar refused it and the refusal above was

bool

sent instead.

Example::

if request.method == "REGISTER":
    if not await 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)
    if not registrar.save(request, flow_token=token):
        return          # the refusal is already answered
    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:

  1. 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.
  2. 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.
  3. No 200 OK is generated — the proxy will relay the upstream's response itself.

Like :meth:save, the cached binding is stamped with :attr:~siphon_sdk.types.Contact.received from the REGISTER's source address.

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,
    include_as_contacts: bool = False,
) -> str

Generate RFC 3680 reginfo XML for an AoR.

Returns the XML document as a string. Lists what the user registered: AS-side capability records (from :meth:save_as_contact) are excluded by default, because RFC 3680 §5.2 defines <contact> as a contact registered for the address of record and an AS that answered a third-party REGISTER has not registered one (TS 24.229 §5.4.1.7). Emitting them showed the UE contacts against its own IMPU that it never created. The iFC-matched capability set reaches the UE through RFC 6809 Feature-Caps on the REGISTER 200 OK instead.

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
include_as_contacts bool

also emit AS capability records as <contact> elements, with their RFC 3840 feature tags as <unknown-param> children (RFC 3680 §5.3.2). Default False — only for a watcher known to want them, never for a UE.

False

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 and restore the default limits (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" (a siphon extension, not a 3GPP Annex H transform).

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