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
|
None
|
max_contacts
|
Optional[int]
|
Bindings one AoR may hold; a new binding past it is
answered |
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:
Expiresbelowmin_expires:423 Interval Too Briefwith aMin-Expiresheader (RFC 3261 §10.3 step 7).- A new binding past
max_contacts:503 Service UnavailablewithRetry-Afterset to the seconds until the soonest held binding expires (at least 1, at mostmax_expires, andmax_expireswhen 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 |
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 |
None
|
flow_token
|
Optional[str]
|
Opaque proxy-side token to attach to every
contact saved by this call. Captures the inbound flow
so subsequent |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
|
bool
|
|
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:
- The contact lifetime is read from the reply's
Expiresheader (the registrar's grant per RFC 3261 §10.3 step 8), not the request's (the UE's ask). UEs commonly ask for600000s; 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_expirescap 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.
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 |
required |
aliases
|
Optional[list[str]]
|
IMS implicit registration set, same shape as
:meth: |
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
when the reply has no parseable |
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
¶
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: |
required |
Returns:
| Type | Description |
|---|---|
list[Contact]
|
List of UE-side :class: |
list[Contact]
|
(descending). Empty list if no UE contacts registered. |
lookup_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: |
required |
Returns:
| Type | Description |
|---|---|
list[Contact]
|
List of matching UE-side :class: |
list[Contact]
|
q-value (descending). AS-side capability records are |
list[Contact]
|
excluded, matching :meth: |
list[Contact]
|
has that contact. |
save_as_contact
¶
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 |
required |
expires_secs
|
Optional[int]
|
lifetime for the cached AS contact. When
|
None
|
Returns:
| Type | Description |
|---|---|
bool
|
|
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
¶
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
|
required |
Returns:
| Type | Description |
|---|---|
Optional[Contact]
|
The matching :class: |
Optional[Contact]
|
or |
is_registered
¶
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: |
required |
is_registered_contact
¶
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: |
required |
aor_count
async
¶
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
¶
Force-expire all contacts for a URI.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
Union[str, SipUri]
|
AoR to expire. |
required |
remove
¶
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 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 contacts (IMS: SAR succeeded).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
Union[str, SipUri]
|
AoR to confirm. |
required |
asserted_identity
¶
Look up stored P-Asserted-Identity for a URI.
Returns:
| Type | Description |
|---|---|
Optional[str]
|
Identity string if stored, otherwise |
set_asserted_identity
¶
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
¶
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
¶
Get stored Service-Route headers for a URI (RFC 3608).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
Union[str, SipUri]
|
AoR as string or :class: |
required |
Returns:
| Type | Description |
|---|---|
list[str]
|
List of Route URI strings, or empty list. |
set_associated_uris
¶
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
¶
Get stored P-Associated-URI list for a URI.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
Union[str, SipUri]
|
AoR as string or :class: |
required |
Returns:
| Type | Description |
|---|---|
list[str]
|
List of P-Associated-URI strings, or empty list. |
on_change
staticmethod
¶
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. |
required |
state
|
str
|
|
'full'
|
version
|
int
|
reginfo version counter (default 0). |
0
|
include_as_contacts
|
bool
|
also emit AS capability records as
|
False
|
Returns:
| Type | Description |
|---|---|
str
|
XML string conforming to RFC 3680. |
add_contact
¶
Add a contact binding directly (test helper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aor
|
str
|
Address-of-record string (e.g. |
required |
contact
|
Contact
|
:class: |
required |
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 |
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. |
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
¶
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
¶
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):
...