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)
Contact: * beside another Contact entry (on the same line or another), or with an Expires that is absent or not zero 400 Bad Request (RFC 3261 §10.3 step 6); only Contact: * alone with Expires: 0 removes every binding

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

registrar.save_proxy() caches what the registrar of record granted, read from its 2xx (RFC 3261 §10.3 step 8): each contact gets the expires parameter on the reply's Contact for that binding (matched by URI), else the reply's Expires header. A contact the reply gives expires=0 is removed, so the 2xx to a de-REGISTER that carries only Contact: <...>;expires=0 clears the binding. It raises ValueError, caching nothing, only when a contact has neither.

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: * with Expires: 0 removes every binding; a * alongside another Contact entry, or with an Expires other than zero or none at all, is answered 400 Bad Request and returns False with the bindings untouched (RFC 3261 §10.3 step 6). 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

restore

restore(
    aor: str,
    contact: str,
    *,
    expires: int,
    path: Optional[list[str]] = None,
    call_id: Optional[str] = None,
    cseq: Optional[int] = None,
    auth_user: Optional[str] = None,
    aliases: Optional[list[str]] = None
) -> bool

Rebuild a binding no REGISTER reached this node for: IMS restoration (TS 23.380 §4.3.2, §4.4.2), from the SCSCF-Restoration-Info the HSS returned.

Each Contact in contact is bound to aor for expires seconds (capped at max_expires), with its +sip.instance, reg-id and other parameters, the RFC 3327 path, and auth_user as the identity that authenticated it. aliases is the implicit registration set, as for :meth:save. call_id and cseq are what the registrar compares a later refresh against; the mock keeps no CSeq state and accepts them.

Returns:

Type Description
bool

True when the binding was stored, False when the registrar

bool

refused it (an AoR that is not a safe storage key).

Raises:

Type Description
ValueError

contact is not a Contact header field value, or expires is 0.

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. Each contact's lifetime is the reply's grant (RFC 3261 §10.3 step 8), not the request's (the UE's ask): the expires parameter on the reply's Contact for that binding (matched by URI), else the reply's Expires header. A Contact the reply gives expires=0 is removed, so the 2xx to a de-REGISTER that carries only Contact: <...>;expires=0 clears the binding. 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 with no Contact expires parameter clears the binding too (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 each Contact's expires parameter and the Expires header).

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 gives a contact no expiration at all: no expires parameter on its Contact and no Expires header. Nothing is cached then.

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 compared case-insensitively, and port; URI parameters and default ports are ignored). siphon answers it from an index, so its cost does not grow with the number of registrations.

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

remove_contact

remove_contact(
    aor: Union[str, SipUri],
    contact: Union[str, SipUri, Contact],
) -> bool

Remove one binding of an AoR, leaving its other contacts registered.

For a reg-event NOTIFY (RFC 3680) or a Cx RTR that ends one contact of a multi-contact AoR, where :meth:remove would drop them all. Covers bindings stored by :meth:save and :meth:save_proxy alike.

contact matches a UE binding the way a REGISTER's expires=0 matches the binding it removes: by the stored Contact URI, angle brackets allowed, so the contact.uri a :meth:lookup reported always matches. AS capability records are never matched. The removal is written through to the registrar backend and fires @registrar.on_change with "deregistered". Removing the last binding removes the AoR as :meth:remove does, its AS records, Service-Route, P-Asserted-Identity, P-Associated-URI and aliases with it::

# a reg-event NOTIFY reports one <contact state="terminated">
if not registrar.remove_contact(aor, contact_uri):
    log.warn(f"no binding of {aor} at {contact_uri}")

Parameters:

Name Type Description Default
aor Union[str, SipUri]

The AoR (or one of its implicit-set aliases).

required
contact Union[str, SipUri, Contact]

The Contact URI to remove, as string, :class:SipUri or a :class:Contact from :meth:lookup.

required

Returns:

Type Description
bool

True when a binding was removed, False when none matched.

network_deregister async

network_deregister(
    aor: Union[str, SipUri],
    source: Optional[tuple[str, int]] = None,
    reason: str = "network-initiated",
) -> int

End a registration on the network's word, as a P-CSCF does when it learns the UE's access is gone (3GPP TS 24.229 clause 5.2.8), e.g. the PCF terminating the AF session of the UE's signalling path.

Each UE binding of aor registered from source ((ip, port)), or every one when source is None, goes: siphon de-registers a save_proxy() binding at the S-CSCF with a REGISTER (Expires: 0) over its Service-Route, fires @registrar.on_change with "deregistered" and shortens or removes the UE's security associations. The mock removes the bindings and records each in :attr:network_deregistrations::

gone = await registrar.network_deregister(aor, (ip, port), "access lost")

Returns:

Type Description
int

How many bindings were de-registered.

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", "expired", or "snapshot" - contacts: list[Contact] — current contact bindings

A handler that also takes a sequence keyword (named, or through **kwargs) is passed it: a per-process number, one higher on every change, the same number the control plane's RegistrationChanged carries as seq. Handlers can overlap (one still running past the engine's 10 s event-handler window), so a script mirroring state keeps the highest number per AoR. A three-argument handler is called as before. The number restarts with the process; see registrar.instance_epoch::

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

"snapshot" is not a change: when the handler fell so far behind the registrar that events were lost (a registration storm against a slow handler), siphon re-reports every registered AoR with its current contacts under "snapshot", so a script that keeps its own copy is whole again. An AoR whose bindings went away among the lost events is not in the snapshot. A handler that only reacts to changes can ignore it.

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, and the soft-UE case of one process registering many IMS UEs, each from its own address (local_addr).

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']}")

Many UEs, each from its own address::

# One entry per UE. The protected ports may repeat across entries:
# the SAs are told apart by address.
registration.add(
    "sip:001010000000001@ims.mnc01.mcc001.3gppnetwork.org",
    "sip:198.51.100.10:5060",
    user="001010000000001@ims.mnc01.mcc001.3gppnetwork.org",
    auth="aka", k=k, opc=opc,
    ipsec=True, ue_port_c=6100, ue_port_s=6101,
    local_addr="192.0.2.77",
    refresh_window=(0.4, 0.6),   # spread refreshes over the fleet
)
registration.set_local_addr(impu, "192.0.2.78")   # the UE moved
aor = registration.find_by_local(call.flow.local_addr)  # MT: which UE?
log.info(str(registration.counts()))

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,
    ipsec_offer: Optional[list[tuple[str, str]]] = None,
    imei: Optional[str] = None,
    ims_features: Optional[list[str]] = None,
    register_body: Optional[
        tuple[str, Union[bytes, str]]
    ] = None,
    register_headers: Optional[dict[str, str]] = None,
    local_addr: Optional[str] = None,
    require_local_addr: bool = False,
    refresh_window: Optional[tuple[float, float]] = None
) -> None

Add a new outbound registration (re-adding an AoR replaces it).

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" (case-insensitive). Anything else raises ValueError: siphon dials no outbound SCTP association and no WebSocket, so a trunk naming one could never register.

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]

Preferred integrity algorithm — "hmac-sha-1-96" (default), "hmac-md5-96", or "hmac-sha-256-128" (a siphon extension, not a 3GPP Annex H transform). Without ipsec_offer the Security-Client offers it first, then "hmac-sha-1-96" and "hmac-md5-96", each with ipsec_ealg, so the defaults offer exactly two mechanisms (sha-1 and md5, both ealg=null).

None
ipsec_ealg Optional[str]

Encryption algorithm paired with those: "null" (default), "aes-cbc" or "des-ede3-cbc".

None
ipsec_offer Optional[list[tuple[str, str]]]

The Security-Client offer, explicitly: a non-empty list of (alg, ealg) pairs, one ipsec-3gpp mechanism each, in the order given (3GPP TS 33.203 clause 6.1, RFC 3329). [("hmac-sha-1-96", "aes-cbc")] offers one mechanism; [("hmac-sha-1-96", "null"), ("hmac-md5-96", "null")] is the default offer. alg is "hmac-sha-1-96", "hmac-md5-96" or "hmac-sha-256-128"; ealg is "null", "aes-cbc" or "des-ede3-cbc". A pair may be a tuple or a two-item list. Needs ipsec=True and replaces ipsec_alg / ipsec_ealg (naming both is an error).

None
imei Optional[str]

IMEI for the Contact +sip.instance (urn:gsma:imei), e.g. "01234567-000001-0". None: no instance tag.

None
ims_features Optional[list[str]]

Contact feature tags to advertise: any of "mmtel", "video", "smsip", "mcptt" (case-insensitive; anything else raises ValueError). "mcptt" adds +g.3gpp.mcptt and the MCPTT ICSI to the Contact (3GPP TS 24.379 clause 7.2.1); with "mmtel" both ICSIs share one +g.3gpp.icsi-ref parameter ("...icsi.mmtel,...icsi.mcptt", TS 24.229 clause 7.9.2). siphon adds no P-Preferred-Service to a REGISTER (TS 24.379 7.2.1 asks for none); pass one in register_headers if the core wants it.

None
register_body Optional[tuple[str, Union[bytes, str]]]

(content_type, body) carried by every REGISTER that registers: the initial one, each refresh, the answer to a 401 (digest or AKA, protected or not) and the protected re-REGISTER. body is bytes or str (UTF-8); Content-Type and Content-Length are set from it. The de-REGISTER (Expires 0) carries none: TS 24.379 7.2.1 asks for it to authorise a binding, not to remove one. The MCPTT service authorisation goes here as ("application/vnd.3gpp.mcptt-info+xml", xml).

None
register_headers Optional[dict[str, str]]

{name: value} added, in order, to the same REGISTERs as register_body. A header siphon builds on a REGISTER (Via, From, To, Call-ID, CSeq, Contact, Expires, Max-Forwards, Route, Content-Type, Content-Length, Authorization, Security-Client/Verify, Require, Supported, Allow, User-Agent, ...) raises ValueError, as does a name that is not a header field name or a value with a line break.

None
local_addr Optional[str]

The IP this entry registers from (an address, no port): the Via/Contact host, the UDP source, and with ipsec the UE side of its SAs. siphon opens the sockets itself, so the address need not exist on the host yet. transport "udp" or "tcp": over TCP siphon listens on ue_port_s (the plain port without ipsec) for the P-CSCF's connection and connects from ue_port_c. Over UDP with ipsec it also accepts TCP on ue_port_s while an SA set stands (the P-CSCF's over-MTU requests), and with listen.mtu set sends a REGISTER over the SA set that exceeds mtu - 200 over TCP from ue_port_c (UDP if the connect fails). None (default): the process's own address.

None
require_local_addr bool

Never register from the process's address. Without a local_addr the entry waits in state "awaiting_address" until :meth:set_local_addr gives it one. Implied by local_addr.

False
refresh_window Optional[tuple[float, float]]

(low, high) fractions of the granted Expires to refresh within, drawn per refresh, with 0 < low <= high < 1. Must be a tuple. None: the registrant default (half the grant).

None

Raises:

Type Description
ValueError

when ipsec_offer is empty, names an unknown alg or ealg or the same pair twice, is given without ipsec=True, or comes with ipsec_alg / ipsec_ealg; when ims_features names an unknown tag; when register_headers names a header siphon owns on a REGISTER or register_body's content type is not a MIME type; when transport is not udp/tcp/tls; when auth="aka" but k or an operator key (op/opc) is missing; when ipsec=True without auth="aka" / ue_port_c / ue_port_s; when local_addr is not an IP address, or is given with transport tls (udp and tcp take one); when another AoR already registers from the same (local_addr, ue_port_c) (ue_port_c is None without ipsec, so two non-ipsec entries cannot share an address); when refresh_window breaks 0 < low <= high < 1; or once the process is shutting down (it de-registered everything and registers nothing again) — mirroring the Rust binding.

TypeError

when refresh_window is not a tuple of two numbers, or ipsec_offer is not a list of (alg, ealg) str pairs.

Example::

# A UE whose address is not known yet: it waits, never
# registering from the process's own address.
registration.add(impu, "sip:198.51.100.10:5060", user=impi,
                 auth="aka", k=k, opc=opc,
                 ipsec=True, ue_port_c=6100, ue_port_s=6101,
                 require_local_addr=True)
assert registration.status(impu) == "awaiting_address"
registration.set_local_addr(impu, "192.0.2.77")

# An MCPTT client: feature tags in the Contact and the service
# authorisation in the body of every REGISTER.
registration.add(impu, "sip:198.51.100.10:5060", user=impi,
                 auth="aka", k=k, opc=opc,
                 ims_features=["mmtel", "mcptt"],
                 register_body=("application/vnd.3gpp.mcptt-info+xml",
                                mcptt_info_xml),
                 register_headers={"P-Preferred-Service":
                     "urn:urn-7:3gpp-service.ims.icsi.mcptt"})

set_local_addr

set_local_addr(aor: str, local_addr: Optional[str]) -> bool

Move a registration to another address, or take its address away.

Parameters:

Name Type Description Default
aor str

The registration's AoR.

required
local_addr Optional[str]

The IP it registers from now (no port), or None for none.

required

Returns:

Type Description
bool

False when there is no such registration, else True.

Restating the current address changes nothing (returns True, state untouched), so this can run on a timer for every UE. A real move starts the entry over: state "unregistered" (due now), failure count, Service-Route and P-Associated-URI cleared, and from then on the entry requires an address of its own. None leaves an entry that has had an address (or was added with require_local_addr=True) in "awaiting_address", firing @registration.on_change with that event; any other entry falls back to the process's address as "unregistered".

Raises:

Type Description
ValueError

local_addr is not an IP address; another AoR already registers from it with the same ue_port_c; the entry's transport is tls (udp and tcp take an address); or the process is shutting down and the address would change (a move starts a fresh registration).

Example::

@timer.every(seconds=30)
def follow_addresses():
    for impu, address in current_ue_addresses().items():
        registration.set_local_addr(impu, address)  # None: detached

find_by_local

find_by_local(local: str) -> Optional[str]

The AoR registering from an address, for routing an MT request to the UE it arrived for.

Parameters:

Name Type Description Default
local str

"ip", "ip:port" or "[v6]:port", e.g. call.flow.local_addr. With a port: the entry on that address owning that port (its plain port, the first UDP listener port, 5060 in the mock; or its ue_port_c / ue_port_s). Without a port: the first entry on the address.

required

Returns:

Type Description
Optional[str]

The AoR, or None when no entry registers from there.

Raises:

Type Description
ValueError

local is not an IP address or IP:port.

Example::

@b2bua.on_invite
def on_invite(call):
    impu = registration.find_by_local(call.flow.local_addr)
    if impu is None:
        call.reject(404, "Not Found")
        return
    call.dial(route_for(impu))

counts

counts() -> dict[str, int]

How many registrations are in each state.

Returns:

Type Description
dict[str, int]

A dict with every state as a key ("unregistered",

dict[str, int]

"registering", "challenging", "registered",

dict[str, int]

"failed", "awaiting_address"), zero when none is in it.

Example::

@timer.every(seconds=60)
def report():
    states = registration.counts()
    ue_registered.set(states["registered"])
    ue_waiting.set(states["awaiting_address"])

remove

remove(aor: str) -> bool

Remove an outbound registration by AoR.

refresh

refresh(aor: str) -> bool

Force an immediate re-registration for an AoR.

Returns:

Type Description
bool

False when there is no such registration, and for every AoR once

bool

the process is shutting down (no registration starts then).

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,

list[dict]

local_addr (str, or None on the process's address),

list[dict]

failure_count (int) and last_failure_status (int, or

list[dict]

None once an attempt succeeded).

Example::

for reg in registration.list():
    if reg["state"] == "failed":
        log.warn(f"{reg['aor']} from {reg['local_addr']}: "
                 f"{reg['last_failure_status']} "
                 f"x{reg['failure_count']}")

status

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

Get the state of a specific registration.

Returns:

Type Description
Optional[str]

One of :attr:STATES (including "awaiting_address"), or None

Optional[str]

when the AoR is unknown.

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: Optional[str] = None
) -> "Optional[Flow]"

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

Parameters:

Name Type Description Default
aor str

The registration's AoR.

required
ue_ip Optional[str]

siphon's own address on the SA (the IP the protected ports are bound to). Defaults to the entry's local_addr; required for an entry without one.

None

Returns:

Type Description
'Optional[Flow]'

The Flow, or None for an unknown AoR. Real runtime also returns

'Optional[Flow]'

None until the sec-agree handshake completes; the mock returns a

'Optional[Flow]'

Flow whenever the entry was added with ipsec=True (so MO

'Optional[Flow]'

handlers can be unit-tested), else None.

Raises:

Type Description
ValueError

ue_ip is not an IP address, or is omitted for an entry without a local_addr.

Example::

flow = registration.flow(impu)   # the entry's own local_addr
if flow is None:
    call.reject(503, "Service Unavailable")
else:
    call.dial(target, flow=flow)

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", "deregistered", "awaiting_address" (the entry needs an address of its own and has none: added with require_local_addr=True and no local_addr, or its address taken away with set_local_addr(aor, None)), or "snapshot" - state: dict -- {"expires_in": int, "failure_count": int, "registrar": str, "state": str | None, "sequence": int, "status_code": int} (sequence is a per-process number, one higher on every change, the seq of the control plane's RegistrantChanged; keep the highest per AoR, since handlers can overlap) (status_code present when event_type is "failed", and on a "snapshot" of a registration whose last attempt failed; state is the registration's current state, "unregistered", "registering", "challenging", "registered", "failed" or "awaiting_address", None once it was removed)

"snapshot" is not a change: when the handler fell so far behind that events were lost, siphon re-reports every outbound registration under "snapshot", with its current state.

Usage::

@registration.on_change
def on_trunk_change(aor, event_type, state):
    if event_type == "awaiting_address":
        log.warn(f"{aor} has no address to register from")

clear

clear() -> None

Reset all registrations (test helper).

One process, many IMS UEs

An outbound registration can register from an address of its own instead of the process's. That is what lets one process act as many IMS UEs, each from its own address. The walkthrough is the IMS multi-UE cookbook.

registration.add(
    "sip:001010000000001@ims.mnc01.mcc001.3gppnetwork.org",
    "sip:pcscf.ims.mnc01.mcc001.3gppnetwork.org:5060",
    user="001010000000001@ims.mnc01.mcc001.3gppnetwork.org",
    auth="aka", k=k, opc=opc,
    ipsec=True, ue_port_c=6100, ue_port_s=6101,
    imei="01234567-000001-0", ims_features=["mmtel", "smsip"],
    local_addr="203.0.113.11",      # IP only
    require_local_addr=True,        # never fall back to the process's address
    refresh_window=(0.4, 0.6),      # overrides registrant.refresh_window
)
Argument / method Meaning
add(..., local_addr=None) The IP this entry registers from: Via and Contact host, the UDP source of every REGISTER (initial, 401 retry, de-REGISTER), and with ipsec=True the UE side of its SAs, the protected REGISTER leaving from local_addr:ue_port_c. siphon opens the sockets itself (IP_FREEBIND), so the address need not exist yet and no listen.udp entry is needed. UDP or TCP: over TCP siphon listens on local_addr:ue_port_s (the plain port without IPsec) for the P-CSCF's connection and connects from local_addr:ue_port_c (the plain port for the unprotected initial REGISTER).
add(..., require_local_addr=False) True: never register from the process's address. Without local_addr the entry waits as awaiting_address.
add(..., refresh_window=None) (low, high) fractions of the granted Expires, drawn per refresh; 0 < low <= high < 1. None: registrant.refresh_window.
set_local_addr(aor, ip_or_None) -> bool Move the entry. Restating the current address is a no-op, so it can run on a timer. A real move deletes the old SA pair and starts a fresh registration from the new address at once, keeping the Call-ID. None puts an entry that has had an address (or has require_local_addr) in awaiting_address; any other entry falls back to the process's address. False when there is no such AoR.
find_by_local("ip" \| "ip:port") -> str \| None The AoR registering from that address, for MT routing: pass call.flow.local_addr or request.flow.local_addr. With a port, the entry owning that port (plain, protected client or protected server); without, the first entry on the address. A linear scan, meant for once per call.
counts() -> dict Registrations per state, every state present (zero when empty).
list() Each dict also carries local_addr (str or None), failure_count and last_failure_status (int or None once an attempt succeeded).
flow(aor, ue_ip=None) ue_ip defaults to the entry's local_addr; it is required (ValueError) for an entry on the process's address.

add and set_local_addr raise ValueError when another entry already registers from the same (local_addr, ue_port_c), for a local_addr on a tls entry, and for a malformed window.

States are unregistered, registering, challenging, registered, failed and awaiting_address. @registration.on_change gets awaiting_address as an event_type when an entry loses its address.

Limits:

  • UDP or TCP, not TLS. Over TCP the UE's own connection to the P-CSCF leaves from its protected client port (TS 24.229 5.1.1.2.2) and the P-CSCF's connection arrives on a per-UE listener on its protected server port (5.2.2.2), answered on that connection. Moving, removing and shutdown close both. The UE side has no TLS identity, so tls is refused.
  • A move does not de-register the old binding. The old address is usually gone, so nothing could carry the de-REGISTER. A registrar that keys on +sip.instance (RFC 5626, built from imei) replaces the binding with the new one; any other lets it lapse at its Expires.
  • Five sockets per IPsec UE (plain, three protected client ports, protected server), so the descriptor limit has to cover five per UE plus the base set.
  • The Contact port is the plain UDP port, the first listen.udp port, as for a single UE.
  • Identity overlay: an MT request that arrives on a per-UE socket is answered with that socket's own address in Contact and Via, not advertised_address, and the per-UE addresses count as this node when Route entries are consumed.

Protected re-registration (IMS UE)

A registered IPsec entry re-registers and de-registers over the SA set it holds (TS 24.229 5.1.1.4, 5.1.1.6): from its protected client port to the P-CSCF's protected server port, with the Authorization of its last authentication, the Security-Server of that 401 in Security-Verify, and new SA parameters in Security-Client. The Via and the Contact name the protected server port (TS 24.229 5.1.1.4.2 b, c), as does the authenticated REGISTER after any 401: that is the binding the S-CSCF keeps, so MT requests come over the set to the protected server port. Only the unprotected initial REGISTER names the plain port.

  • Not challenged: the 200 extends the set's lifetime to the grant plus 30 s, never shortening it (TS 24.229 5.1.1.4.2). No new SA.
  • Challenged: the new set is installed beside the old one on another protected client port (TS 33.203 7.4.0: the client port changes, the server port stays) and the authenticated REGISTER rides it. Until its 200 the new set lives 240 s, the reg-await-auth timer (TS 24.229 5.1.1.5.1); at the 200 it takes over, for the grant plus 30 s or what the old set had left if that is longer. The old one stays for the P-CSCF until the answer to the next REGISTER arrives over the new one, then goes (TS 33.203 7.4.1a). A P-CSCF that keeps its protected client port (siphon's own does) gives both sets one selector toward the UE's server port; the UE's policies there accept either set's SA, so a request the P-CSCF still sends over the old set during the overlap is received. Policies are re-pinned with the states, so no policy ends under a live set.
  • Invalid challenge: an SQN out of range (AUTS) or a bad MAC is answered over the established set, or unprotected without one, with a new Security-Client and no temporary set (TS 24.229 5.1.1.5.3). A third consecutive invalid challenge ends the attempt (5.1.1.5.12).
  • Failed or unanswered (Timer F): the new set, if any, goes and the next REGISTER is an unprotected one (TS 24.229 5.1.1.4.1). Its authentication deletes the set kept from before. A set the kernel no longer holds also sends the next REGISTER unprotected.
  • remove(): the protected de-REGISTER goes out and the SAs are deleted when it is answered, or after Timer F.
  • Shutdown: every entry is removed as remove() does, and the sets are deleted before the process exits: as each de-REGISTER is answered, and 4 s after they were sent for the rest, deletions bounded to 5 s more (TS 33.203 7.4.1a). Nothing is left to the hard lifetime.
  • A protected port that will not bind: a rotation port no set holds moves to another free port; the established set's own client port takes that set out of use, so the next REGISTER goes unprotected. Each is logged and counted in siphon_ipsec_ue_port_bind_failures_total. Only an unbindable plain or server port, or no client port left, holds the entry back.
  • SPIs: an offer never reuses an SPI of a set the UE holds, of a pair the IPsec manager owns, or of a kernel state present when the loop started, one a previous process left included (TS 33.203 7.1).

The UE rotates through three protected client ports: ue_port_c and the two above the higher of ue_port_c and ue_port_s (6100/6101 uses 6100, 6102 and 6103). Three, because it holds at most two sets at rest and a P-CSCF refuses an offered client port one of its sets still uses (TS 33.203 7.1). siphon opens them itself, on the entry's address or the process's. A protected port of 0, 5060 or 5061, or two entries on one address sharing any protected port, is refused.

Not covered: the registrant does not subscribe to the reg event package, so a network-initiated de-registration (TS 24.229 5.1.1.7) is learned from the next refresh's answer, which starts an initial registration, rather than from a NOTIFY.