SIP types

Small value objects that flow through the scripting API: parsed URIs, contact bindings, the captured inbound flow, and the action record the test harness uses to report what a handler did.

SipUri

A parsed SIP / SIPS / tel URI. Reachable via request.ruri, request.from_uri, request.to_uri, and Contact.uri parsing.

A parsed SIP or SIPS URI (e.g. sip:alice@example.com:5060).

At runtime this is backed by the Rust PySipUri class. The mock version is a plain dataclass with the same properties.

Examples::

uri = SipUri(user="alice", host="example.com")
assert str(uri) == "sip:alice@example.com"

uri = SipUri(scheme="sips", host="proxy.example.com", port=5061)
assert str(uri) == "sips:proxy.example.com:5061"

scheme class-attribute instance-attribute

scheme: str = 'sip'

URI scheme — "sip" or "sips".

user class-attribute instance-attribute

user: Optional[str] = None

User part of the URI (e.g. "alice"). None for server URIs.

host class-attribute instance-attribute

host: str = 'localhost'

Host or domain (e.g. "example.com", "10.0.0.1").

port class-attribute instance-attribute

port: Optional[int] = None

Port number. None means default (5060 for SIP, 5061 for SIPS).

is_local property

is_local: bool

True if the host matches one of the locally configured domains.

In the mock, set via SipUri._is_local = True or via the test harness local_domains parameter.

is_tel property

is_tel: bool

True if the scheme is tel:.

Contact

A registered contact binding returned by registrar.lookup(...).

A registered contact binding returned by registrar.lookup().

Attributes:

Name	Type	Description
uri	str	The contact URI string (e.g. "sip:alice@192.168.1.5:5060").
received	Optional[str]	Source address of the REGISTER as a SIP URI, or None.
q	float	Quality value between 0.0 and 1.0 (higher = preferred).
expires	int	Seconds remaining until this binding expires.
age_secs	int	Seconds since the binding was created or last refreshed.

uri instance-attribute

uri: str

Contact URI as a string.

received class-attribute instance-attribute

received: Optional[str] = None

Transport source address of the REGISTER that created this binding, as a SIP URI — "sip:<ip>:<port>;transport=<proto>", the shape OpenSIPS puts in its received_avp. None when the binding was saved without source address info.

Prefer this over :attr:uri when routing: a Contact URI can carry a private or NATed address, while this is the address the UE is actually reachable on::

request.fork([c.received or c.uri for c in contacts])

Behind NAT the two differ — the UE advertises sip:alice@10.0.0.5:5060 in its Contact while the packet arrives from 198.51.100.20:41234 — and only the latter is routable. It is a full URI rather than a bare host:port precisely so it can be handed to fork() / relay() verbatim.

q class-attribute instance-attribute

q: float = 1.0

Quality value (0.0–1.0). Higher values are preferred. Default is 1.0 per RFC 3261.

expires class-attribute instance-attribute

expires: int = 3600

Seconds remaining until this contact binding expires.

age_secs class-attribute instance-attribute

age_secs: int = 0

Seconds since this binding was created or last refreshed.

Use for recency rules — :attr:expires cannot answer that question, because it is time remaining and every UE asks for a different lifetime: a handset that requested 600 s and registered a second ago sorts below one that requested 3600 s an hour ago.

registrar.lookup() already returns bindings most-recent-first within one q-value; this is for scripts that need a different rule::

fresh = [c for c in registrar.lookup(uri) if c.age_secs < 3600]

Monotonic (immune to wall-clock steps) and preserved across a restart for bindings restored from a persistence backend. A binding whose stored record pre-dates age tracking reports 0.

path class-attribute instance-attribute

path: list = field(default_factory=list)

RFC 3327 Path headers stored with this binding.

Passing the Contact itself to request.fork() makes the proxy build this branch's Route header set from the Path automatically (in order, per RFC 3327 §5.3) and route the branch by its topmost entry — which is what lets the bindings of one AoR fail over independently, since two bindings usually carry different Path tokens. Read it directly only when routing by hand with request.set_header("Route", ...).

instance_id class-attribute instance-attribute

instance_id: Optional[str] = None

Stable identity of the siphon instance that originally accepted the REGISTER (typically the StatefulSet pod name). None for legacy bindings or deployments that do not configure server.instance_id.

instance_epoch class-attribute instance-attribute

instance_epoch: Optional[str] = None

Boot-time epoch UUID of the process that accepted the REGISTER. Combined with :attr:instance_id, distinguishes successive runs of the same logical replica.

is_local class-attribute instance-attribute

is_local: bool = False

True when the binding's (instance_id, instance_epoch) matches the current siphon process — i.e. this process accepted the REGISTER. Useful for graceful-shutdown deregister and NAT keepalive ownership.

flow_token class-attribute instance-attribute

flow_token: Optional[str] = None

Opaque proxy-side token attached at REGISTER time via registrar.save(flow_token=...). None for non-P-CSCF bindings.

flow class-attribute instance-attribute

flow: Optional[Flow] = None

Captured inbound flow (Flow view). Pass to request.relay(flow=...) / request.fork(contacts) / call.dial(flow=...) for RFC 5626 §5.3 connection reuse — the only way to reach a WebSocket UE (RFC 7118 §5). Populated for any binding this process accepted (no flow_token= required); None only for a binding restored cross-instance whose local listener / connection id aren't available here. Guard on :attr:is_local before routing over it.

params class-attribute instance-attribute

params: list = field(default_factory=list)

Contact-header parameters preserved from the originating REGISTER (or 3PR 200 OK), excluding tag, q, expires, +sip.instance, and reg-id which are broken out into other fields. Each entry is a (name, value) tuple — value is None for flag parameters (e.g. +g.3gpp.smsip) and a string for valued parameters (e.g. +g.3gpp.icsi-ref="urn:...").

Surfaced verbatim by :func:registrar.reginfo_xml as <unknown-param> children per RFC 3680 §5.3.2 so watchers see the same capability advertisement the registrar received.

kind class-attribute instance-attribute

kind: str = 'ue'

"ue" (UE-side binding from a REGISTER — the default and the only contacts returned by :func:registrar.lookup) or "as" (application-server capability record captured from a 3PR 200 OK via :func:registrar.save_as_contact). AS contacts surface in reg-event NOTIFY bodies (TS 24.229 §5.4.2.1.2) but are excluded from routing lookups.

client_transport class-attribute instance-attribute

client_transport: Optional[str] = None

Transport the client used to reach the front when this binding was registered ("tls", "wss", …), or None when no front declared one.

Descriptive, like :attr:received: it records that a UE behind a TLS-terminating front spoke TLS even though siphon accepted the REGISTER over plain TCP. Routing still follows :attr:flow and :attr:uri, which name the socket the UE is actually reachable on. None does not mean insecure — it means nothing contradicted the hop the REGISTER arrived on.

auth_user class-attribute instance-attribute

auth_user: Optional[str] = None

Authenticated identity that stored this binding: what request.auth_user held when registrar.save() ran, which in IMS is the private identity (IMPI). None when the REGISTER was saved without an authenticated user, and on every registrar.save_proxy() cache.

auth.verify_integrity_protected() trusts a protected re-/de-REGISTER only when its IMPI is this one, so a binding saved without it is challenged on its next protected refresh::

bindings = registrar.lookup("sip:001010000000001@ims.example.com")
owners = {c.auth_user for c in bindings}

Flow

An opaque view of the inbound flow captured at REGISTER time, used for Path-token MT routing and for RFC 5626 connection reuse.

Flows compare by value and hash, so a call can be authorised by matching it against the connection the registration arrived on, rather than by challenging every INVITE with a 407:

@b2bua.on_invite
def on_invite(call):
    bindings = registrar.lookup(str(call.from_uri))
    if any(contact.flow == call.flow for contact in bindings):
        call.dial(str(call.ruri))       # same connection as the REGISTER
    else:
        call.reject(403, "Forbidden")

Equality covers the transport, both addresses and the connection id together. On a stream transport (TCP/TLS/WS/WSS) that is an exact match on one accepted socket, which is why it is worth doing: a source-address check is worthless behind carrier NAT, where every subscriber on the network shares an address. On UDP there is no connection, so a flow carries no more assurance than the address does — treat it as a hint, not authorisation.

The match survives the UE reusing the connection across many calls: the connection id identifies the socket, not the transaction.

Opaque view of an inbound flow captured at REGISTER time.

Returned by :attr:Contact.flow, :attr:Request.flow and :attr:Call.flow. Pass back to :meth:Request.relay (flow= kwarg) to send a request over the same listener that received the REGISTER — bypassing DNS resolution of the Request-URI. Used by P-CSCF MT routing (RFC 3327 §5 / TS 24.229 §5.2.7.2) where the UE's Contact URI is unreachable (NAT, IPSec) and the only path back is the captured flow.

Flows compare by value and hash, so the RFC 5626 connection-reuse test is written directly::

@b2bua.on_invite
def on_invite(call):
    bindings = registrar.lookup(str(call.from_uri))
    if any(c.flow == call.flow for c in bindings):
        call.dial(str(call.ruri))     # same connection as the REGISTER
    else:
        call.reject(403, "Forbidden")

Equality covers the transport, both addresses and the connection id together. On a stream transport that makes it an exact match on one accepted socket, which is a far stronger signal than a source-address check — worthless behind carrier NAT, where every subscriber on the network shares an address. On UDP there is no connection, so the flow carries no more assurance than the address does.

Treat as opaque: scripts read :attr:is_alive for defensive checks but should not depend on the internal field shapes.

transport class-attribute instance-attribute

transport: str = 'udp'

Lowercase transport name: "udp", "tcp", "tls", "ws", or "wss".

remote_addr class-attribute instance-attribute

remote_addr: str = '0.0.0.0:0'

String form of the captured UE source address ("ip:port").

local_addr class-attribute instance-attribute

local_addr: str = '0.0.0.0:0'

String form of the captured listener local address — load-bearing for IPSec sec-agree where the protected port pair must be preserved (3GPP TS 33.203 §7.4).

connection_id class-attribute instance-attribute

connection_id: int = 0

Identifier of the accepted inbound connection this flow was captured on.

For a stream transport (TCP/TLS/WS/WSS) this identifies one accepted socket, so it is what distinguishes a UE still on the connection its REGISTER arrived on from one that reconnected. For UDP it is a deterministic hash of (local_addr, remote_addr) — there is no connection to speak of.

Prefer comparing whole flows over reading this: equality covers the transport and both addresses as well, and a connection id is only meaningful alongside them.

is_alive property

is_alive: bool

Whether the flow is still usable.

For UDP, always True: the listener socket survives any individual exchange. For stream transports (TCP/TLS/WS/WSS), the real implementation returns True only while the exact accepted connection that delivered the REGISTER is still open on this process — a real lookup against the unified stream-connection registry (see PyFlow.is_alive in src/script/api/registrar.rs). A UE that reconnected, or whose socket closed, reports False.

The mock always returns True (no live connections to track).

Action

The record the test harness captures for each action a handler takes (reply, relay, fork, reject, …). Scripts do not create these; assertions read them.

Records a single action taken by a handler (reply, relay, fork, etc.).

Used by the test harness to capture what a script did in response to a SIP message, so you can assert on it.

kind instance-attribute

kind: str

Action type: "reply", "relay", "fork", "reject", "dial", "terminate", "handover", "record_route", "silent_drop".

status_code class-attribute instance-attribute

status_code: Optional[int] = None

For reply / reject — the SIP status code (e.g. 200, 404).

reason class-attribute instance-attribute

reason: Optional[str] = None

For reply / reject — the reason phrase (e.g. "OK").

next_hop class-attribute instance-attribute

next_hop: Optional[str] = None

For relay — the explicit next-hop URI, or None for default.

targets class-attribute instance-attribute

targets: Optional[list[str]] = None

For fork — list of target URIs.

strategy class-attribute instance-attribute

strategy: Optional[str] = None

For fork — "parallel" or "sequential".

timeout class-attribute instance-attribute

timeout: Optional[int] = None

For dial / fork — timeout in seconds.

headers_set class-attribute instance-attribute

headers_set: dict[str, str] = field(default_factory=dict)

Headers that were set/modified before this action.

headers_removed class-attribute instance-attribute

headers_removed: list[str] = field(default_factory=list)

Headers that were removed before this action.

extras class-attribute instance-attribute

extras: Optional[dict] = None

Additional action-specific data (e.g. session timer params, SRS URI).

reliable class-attribute instance-attribute

reliable: bool = False

For reply — RFC 3262 reliable provisional flag (Require: 100rel).