Skip to content

Request

The Request object is passed to every @proxy.on_request handler. It gives read access to the parsed SIP message and methods to reply, relay, fork, and rewrite headers before forwarding.

from siphon import proxy

@proxy.on_request("INVITE")
def route(request):
    if request.ruri.is_local:
        request.relay()
    else:
        request.reply(403, "Forbidden")

A SIP request message.

This object is passed to @proxy.on_request handlers. It provides read-only access to parsed SIP headers and methods to reply, relay, fork, and manipulate the message before forwarding.

In the real SIPhon engine this is backed by a Rust PyRequest with an Arc<Mutex<SipMessage>> inside. The mock version stores everything in plain Python attributes.

Example::

@proxy.on_request
def route(request):
    if request.method == "REGISTER":
        request.reply(200, "OK")
        return
    request.relay()

method property

method: str

SIP method string (e.g. "INVITE", "REGISTER", "BYE").

ruri property

ruri: SipUri

Request-URI as a :class:SipUri object.

from_uri property

from_uri: Optional[SipUri]

From header URI as a :class:SipUri, or None.

to_uri property

to_uri: Optional[SipUri]

To header URI as a :class:SipUri, or None.

from_tag property

from_tag: Optional[str]

From-tag parameter (always present for outgoing requests).

to_tag property

to_tag: Optional[str]

To-tag parameter. None for initial (out-of-dialog) requests.

call_id property

call_id: Optional[str]

Call-ID header value.

cseq property

cseq: Optional[tuple[int, str]]

CSeq as (sequence_number, method) tuple.

in_dialog property

in_dialog: bool

True if both From-tag and To-tag are present (mid-dialog request).

max_forwards property

max_forwards: int

Max-Forwards header value.

body property

body: Optional[bytes]

Message body (SDP, etc.), or None if empty.

content_type property

content_type: Optional[str]

Content-Type header value (e.g. "application/sdp").

transport property

transport: str

Transport protocol: "udp", "tcp", "tls", "ws", "wss".

source_ip property

source_ip: str

Source IP address of the sender.

source_port property

source_port: int

Source port of the sender.

user_agent property

user_agent: Optional[str]

User-Agent header value.

auth_user property writable

auth_user: Optional[str]

Authenticated username (set after digest auth succeeds).

contact_expires property

contact_expires: Optional[int]

Contact expires value from Contact expires= param or Expires header.

event property

event: Optional[str]

Event header value (e.g. "reg", "presence").

route_user property

route_user: Optional[str]

User part of the top Route header URI, or None.

Reflects the current state of the Route header — after any loose_route() calls have stripped routes addressed to this proxy. For the user-part of a Route the framework consumed (e.g. the IMS service-route's orig/term indicator), use :attr:consumed_route_user.

consumed_route_user property

consumed_route_user: Optional[str]

User part of the first Route entry that loose_route() consumed, or None if no Route was popped yet.

Mirrors the production request.consumed_route_user getter so IMS scripts can read the orig/term user-part of the service-route the P-CSCF preloaded.

consumed_routes property

consumed_routes: list[str]

URIs of all Route entries consumed by loose_route(), in the order they were popped (topmost first).

Empty until the script calls :meth:loose_route.

consumed_route property

consumed_route: Optional[SipUri]

First Route entry consumed by loose_route(), parsed as a :class:SipUri so the script can read user / host / port directly without string-munging.

Returns None when no Route was consumed yet. Convenience over :attr:consumed_route_user for P-CSCF Path-token MT routing where the script needs both the userpart (the opaque token to look up) and the host (to verify it points at this proxy).

flow property

flow

View of the inbound flow this request arrived on, or None for synthetic requests without listener context.

Pass to :meth:relay (flow= kwarg) for Path-token MT routing — sends the request back over the same listener that received the REGISTER without DNS-resolving the URI (RFC 3327 §5 / TS 24.229 §5.2.7.2).

In the mock, returns the test-fixture flow if one was attached via the test harness; otherwise None.

charging_params property

charging_params: list[tuple[str, str]]

List of (name, value) tuples stashed via :meth:set_charging_param. Test helper.

reply_headers property

reply_headers: list[tuple[str, str]]

All (name, value) pairs queued for the response, with replace/add ops resolved (replace wipes earlier values for the same name, add appends). Order preserved per name.

reply_header_ops property

reply_header_ops: list[tuple[str, str, str]]

All (op, name, value) triples as queued — exposes the raw replace/add semantics for tests that need to inspect the queue before resolution.

is_ipsec_protected property

is_ipsec_protected: bool

Whether this request arrived over an IPsec-protected SA.

Mock returns False by default; tests can override the underlying _is_ipsec_protected attribute.

matched_sa property

matched_sa

Handle to the SA that decrypted this request, or None.

actions property

actions: list[Action]

All actions recorded by this request (test-only).

last_action property

last_action: Optional[Action]

The last (most recent) action, or None.

reply

reply(
    code: int, reason: str, reliable: bool = False
) -> None

Send a SIP response.

Parameters:

Name Type Description Default
code int

SIP status code (e.g. 200, 401, 404, 486).

required
reason str

Reason phrase (e.g. "OK", "Not Found").

required
reliable bool

RFC 3262 — when True and code is 101..199, send as a reliable provisional response (Require: 100rel + RSeq). Only honoured for INVITE responses where the UAC advertised 100rel in Supported or Require. Siphon retransmits the response per RFC 3262 §3 (T1 doubling to T2, deadline 32s) until a matching PRACK arrives, then auto-200s the PRACK.

False

Example::

request.reply(200, "OK")
request.reply(486, "Busy Here")
request.reply(183, "Session Progress", reliable=True)

relay

relay(
    next_hop: Optional[str] = None,
    on_reply: Optional[Callable] = None,
    on_failure: Optional[Callable] = None,
    flow: Optional[Flow] = None,
    send_socket: Optional[str] = None,
) -> None

Forward the request to its destination.

Parameters:

Name Type Description Default
next_hop Optional[str]

Optional explicit next-hop URI. If None, the Request-URI is used as the destination.

None
on_reply Optional[Callable]

Optional callback (request, reply) invoked when any response arrives for this relay.

None
on_failure Optional[Callable]

Optional callback (request, code, reason) invoked when an error response (4xx+) arrives.

None
flow Optional[Flow]

Optional :class:Flow (typically binding.flow from registrar.lookup_by_token(...)). When supplied, bypasses DNS resolution of the Request-URI and sends the request directly to the captured inbound flow's listener — load-bearing for P-CSCF MT routing where the UE's Contact URI is unreachable (NAT, IPSec). Ignores next_hop when set.

None
send_socket Optional[str]

Optional egress socket pin ("<transport>:<ip>:<port>", e.g. "udp:10.0.0.1:5060") — the operator equivalent of Kamailio's force_send_socket(). Selects which of siphon's own configured listeners the request leaves from on a multi-homed host. The outgoing Via advertises that listener's address so the response comes back to the same socket. UDP pins the exact (ip, port) listener; TCP/TLS bind the source IP with an ephemeral port. Ignored when flow is set (the flow already pins egress), and when its transport doesn't match the routed transport. A malformed spec raises ValueError; a well-formed spec that names no configured listener is logged and falls back to default routing (never dropped).

None

Example::

request.relay()                           # default routing
request.relay("sip:proxy@10.0.0.2:5060")  # explicit next-hop
request.relay(on_reply=my_reply_handler)   # per-relay callback
request.relay(send_socket="udp:10.0.0.1:5060")  # pin egress NIC
# Path-token MT routing:
binding = registrar.lookup_by_token(token)
request.relay(flow=binding.flow)

fork

fork(
    targets: list[Union[str, Contact]],
    strategy: str = "parallel",
    send_socket: Optional[str] = None,
) -> None

Fork the request to multiple targets.

Parameters:

Name Type Description Default
targets list[Union[str, Contact]]

List of URI strings or :class:Contact objects. Pass Contact objects (not just .uri) so a binding this process accepted (contact.is_local) routes its branch over the captured inbound flow — RFC 5626 §5.3 connection reuse, the only way to reach a WebSocket UE (RFC 7118 §5). Non-local contacts fall back to URI routing.

required
strategy str

"parallel" (all at once, first 2xx wins) or "sequential" (try in q-value order, next on failure).

'parallel'
send_socket Optional[str]

Optional egress socket pin applied to every branch (same "<transport>:<ip>:<port>" form as :meth:relay). A per-branch captured flow still takes precedence over it for that branch.

None

Example::

contacts = registrar.lookup(request.ruri)
# Pass Contact objects so a WebSocket UE routes over its flow.
request.fork(contacts)
request.fork(["sip:a@host", "sip:b@host"], strategy="sequential")
request.fork(contacts, send_socket="udp:10.0.0.1:5060")

record_route

record_route() -> None

Insert a Record-Route header so that subsequent in-dialog requests traverse this proxy.

Must be called before relay() or fork().

loose_route

loose_route() -> bool

Perform RFC 3261 §16.4 loose routing.

If the topmost Route entry carries an lr parameter, pop it, record its URI on :attr:consumed_routes, and return True. If a Route header is present but the topmost entry is a strict route (no lr), return False and leave it intact.

When no Route header is present, the mock falls back to returning :attr:in_dialog to preserve compatibility with scripts that gate loose_route() behind in_dialog checks without setting up explicit Route headers in the test request.

Example::

if request.in_dialog:
    if request.loose_route():
        request.relay()
    else:
        request.reply(404, "Not Here")

get_header

get_header(name: str) -> Optional[str]

Get the first value of a header by name (case-insensitive).

Parameters:

Name Type Description Default
name str

Header name (e.g. "Via", "Contact").

required

Returns:

Type Description
Optional[str]

Header value string or None if not present.

header

header(name: str) -> Optional[str]

Alias for :meth:get_header.

Example::

ua = request.header("User-Agent")

set_header

set_header(name: str, value: str) -> None

Set (replace) a header value.

Parameters:

Name Type Description Default
name str

Header name.

required
value str

New header value.

required

Example::

request.set_header("X-Custom", "my-value")

rewrite_identities

rewrite_identities(
    policy: Optional[str] = None,
    format: Optional[str] = None,
    headers: Optional[list[str]] = None,
    home: Optional[str] = None,
) -> int

Rewrite dialable identity userparts into a target E.164 shape.

Walks From, To, P-Asserted-Identity, P-Preferred-Identity, the Request-URI (and any opted-in Referred-By / Remote-Party-ID), reformatting each dialable number in place. Display names, tags, hosts, non-numbers and preserved service codes are left untouched.

Pass either a named policy from the number_policies: config or an inline format ("e164" | "plain" | "international" | "national") with an optional headers list and a home country-code override. Returns the number of headers changed.

Example::

request.rewrite_identities("teams-outbound@2026")
request.rewrite_identities(format="e164")
request.rewrite_identities(format="national", headers=["From", "request-uri"])

set_charging_param

set_charging_param(name: str, value: str) -> None

Stash a charging-param for the Rf auto-emit hook.

Recognised names (TS 32.299 IMS-Information AVPs):

  • "outgoing-trunk-group-id" — BGCF/MGCF settlement
  • "incoming-trunk-group-id"
  • "application-server" — MMTel-AS
  • "application-provided-called-party-address"

Unknown names are silently kept on the request so future siphon versions can recognise them without breaking scripts. Tests can assert on the captured values via the charging_params property.

Example::

gw = gateway.select("trunks")
request.set_charging_param(
    "outgoing-trunk-group-id", gw.attrs["group"],
)
request.relay(gw.uri)

set_reply_header

set_reply_header(name: str, value: str) -> None

Set (replace) a single-value header on the response built by :meth:reply or :meth:registrar.save.

Use this for single-value headers per RFC 3261 §7.3.1 — To, From, Contact, Expires, Server, Content-Type, Require, Min-Expires. The dispatcher removes any existing header of the same name before inserting this one, so the response will carry exactly one value even if the framework copied a header from the request (e.g. To, From) before the script ran.

For multi-value headers (Via, Route, Service-Route, P-Associated-URI, Path), use :meth:add_reply_header instead.

Parameters:

Name Type Description Default
name str

Header name.

required
value str

Header value.

required

Example::

our_tag = f"scscf-{request.call_id[:8]}"
request.set_reply_header(
    "To", f"{request.get_header('To')};tag={our_tag}"
)
request.reply(200, "OK")

add_reply_header

add_reply_header(name: str, value: str) -> None

Append a header to the response built by :meth:reply or :meth:registrar.save.

Use this for multi-value headers — Via, Record-Route, Route, Service-Route, P-Associated-URI, Path. Multiple calls with the same name accumulate in insertion order, and any existing values copied by the framework (e.g. Via) are preserved.

For single-value headers, use :meth:set_reply_header.

Parameters:

Name Type Description Default
name str

Header name.

required
value str

Header value.

required

Example::

registrar.save(request)
request.add_reply_header(
    "P-Associated-URI", "<sip:user@ims.net>"
)
request.add_reply_header(
    "Service-Route", "<sip:orig@scscf:6060;lr>"
)

set_reply_to_tag

set_reply_to_tag(tag: str) -> None

Attach a To-tag to the response built by :meth:reply.

Required by RFC 3261 §12.1.1.2 / RFC 6665 §4.1.3 on the dialog-establishing 2xx (and any 1xx that establishes early dialog) for INVITE / SUBSCRIBE / REFER from a UAS.

Reads the request's To header, sets or overwrites the ;tag= parameter, and queues the result for replace semantics. Idempotent: calling twice with different tags leaves the most recent tag on the response.

Parameters:

Name Type Description Default
tag str

The To-tag value (no tag= prefix, no quoting).

required

Example::

request.set_reply_to_tag(f"scscf-{request.call_id[:8]}")
request.reply(200, "OK")

get_reply_header

get_reply_header(name: str) -> Optional[str]

Return the reply header value set by :meth:set_reply_header or :meth:add_reply_header, or None if not set.

Test-only convenience — applies replace/add ops in order so the result matches what the dispatcher would inject into the outgoing response. Multi-value (add) entries are joined with ,.

set_body

set_body(
    body: Union[str, bytes], content_type: str | None = None
) -> None

Replace the body of the incoming request message.

Parameters:

Name Type Description Default
body Union[str, bytes]

str or bytes — the new body.

required
content_type str | None

Optional Content-Type to set alongside the body.

None

Example::

request.set_body(pidf_lo_xml, "application/pidf+xml")

set_reply_body

set_reply_body(
    body: Union[str, bytes], content_type: str
) -> None

Attach a body to the response built by :meth:reply.

The dispatcher copies this body and sets Content-Type / Content-Length on the outgoing response.

Parameters:

Name Type Description Default
body Union[str, bytes]

str or bytes — the response body.

required
content_type str

Content-Type header value.

required

Example::

request.set_reply_body(pidf_lo_xml, "application/pidf+xml")
request.reply(200, "OK")

remove_header

remove_header(name: str) -> None

Remove a header entirely.

Parameters:

Name Type Description Default
name str

Header name to remove.

required

has_header

has_header(name: str) -> bool

Check if a header exists (case-insensitive).

Parameters:

Name Type Description Default
name str

Header name.

required

Returns:

Type Description
bool

True if the header is present.

parse_security_client

parse_security_client() -> list

Parse the Security-Client header (RFC 3329 / 3GPP TS 33.203).

Returns a list of :class:MockSecurityOffer objects (one per comma-separated offer). Empty list when the header is absent or no offer parses cleanly.

The mock parser is byte-compatible with the Rust side:

  • Splits on top-level commas (respecting quoted strings).
  • Splits each offer on ;; the first token is the mechanism, the remaining key=value pairs populate the offer.
  • UE address is taken from :attr:source_ip.

has_body

has_body(content_type: str) -> bool

Check if the request has a body matching the given content type.

Parameters:

Name Type Description Default
content_type str

MIME type to match (e.g. "application/sdp").

required

Returns:

Type Description
bool

True if a body is present and Content-Type matches.

ensure_header

ensure_header(name: str, value: str) -> None

Set a header only if it is not already present.

Parameters:

Name Type Description Default
name str

Header name.

required
value str

Value to set if header is missing.

required

remove_headers_matching

remove_headers_matching(prefix: str) -> None

Remove all headers whose name starts with a prefix (case-insensitive).

Mirrors the production request.remove_headers_matching — e.g. request.remove_headers_matching("X-") strips every custom header.

Parameters:

Name Type Description Default
prefix str

Header-name prefix (e.g. "X-").

required

remove_from_header_list

remove_from_header_list(name: str, value: str) -> None

Remove one value from a comma-separated multi-value header.

If the header has values "A, B, C" and you remove "B", the result is "A, C".

Parameters:

Name Type Description Default
name str

Header name.

required
value str

The specific value to remove.

required

set_ruri

set_ruri(value: Union[str, SipUri]) -> None

Replace the entire Request-URI.

Parameters:

Name Type Description Default
value Union[str, SipUri]

New URI as a string or :class:SipUri.

required

set_ruri_user

set_ruri_user(value: Optional[str]) -> None

Set the user part of the Request-URI.

Parameters:

Name Type Description Default
value Optional[str]

New user part, or None to clear.

required

Example::

request.set_ruri_user("bob")

set_ruri_host

set_ruri_host(value: str) -> None

Set the host part of the Request-URI.

Parameters:

Name Type Description Default
value str

New host/domain string.

required

set_from_display

set_from_display(display_name: str) -> None

Rewrite the From header display name.

Parameters:

Name Type Description Default
display_name str

New display name (e.g. "Alice Smith").

required

set_to_display

set_to_display(display_name: str) -> None

Rewrite the To header display name.

Parameters:

Name Type Description Default
display_name str

New display name.

required

set_from_uri

set_from_uri(value: str) -> None

Replace the whole From header URI, preserving the display name and From-tag.

The whole-URI form of :meth:set_from_user. The tag survives — unlike a raw set_header("From", "<sip:...>"), which drops the mandatory From-tag.

Parameters:

Name Type Description Default
value str

New From URI (e.g. "sip:+3120@tenant.example.com:5070").

required

set_to_uri

set_to_uri(value: str) -> None

Replace the whole To header URI, preserving the display name and any To-tag.

The whole-URI form of :meth:set_to_user. The tag survives — unlike a raw set_header("To", "<sip:...>").

Parameters:

Name Type Description Default
value str

New To URI (e.g. "sip:1000@ims.example.org").

required

add_path

add_path(uri: str) -> None

Prepend a Path header (P-CSCF registration path).

Parameters:

Name Type Description Default
uri str

URI to prepend (e.g. "sip:pcscf.ims.example.com;lr").

required

add_pcscf_path

add_pcscf_path(token: str) -> None

Insert a Path header (RFC 3327) of the form <sip:TOKEN@${path_host};lr> where path_host comes from the configured ipsec.path_host.

On the matching mobile-terminating request, the topmost Route will be this same URI; loose_route() consumes it, :attr:consumed_route_user exposes the token, and registrar.lookup_by_token(token) resolves back to the stored binding (TS 24.229 §5.2.7.2).

The mock implementation uses a fixed test path host "pcscf.test" so unit tests can exercise the Path / Route / lookup loop without needing the full siphon config.

prepend_route

prepend_route(uri: str) -> None

Prepend a Route header.

Parameters:

Name Type Description Default
uri str

URI to prepend (e.g. "sip:scscf.ims.example.com;lr").

required

add_contact_alias

add_contact_alias() -> None

Append ;alias to the Contact URI (NAT traversal).

set_contact_uri

set_contact_uri(value: str) -> None

Replace the whole Contact header URI, preserving the display name and any Contact header params (q/expires/…).

In a proxy the Contact is the UA's in-dialog target: rewriting it changes where mid-dialog requests (BYE, re-INVITE) are routed. Fine when the new target is meaningful downstream; a footgun otherwise.

Parameters:

Name Type Description Default
value str

New Contact URI (e.g. "sip:bob@192.0.2.9:5080;transport=tcp").

required

set_contact_user

set_contact_user(value: str) -> None

Rewrite only the userpart of the Contact header URI, leaving the host/port/params intact. An empty string clears the userpart.

Parameters:

Name Type Description Default
value str

New Contact userpart (e.g. an extension), or "" to clear.

required

fix_nated_register

fix_nated_register() -> None

Add received= and rport= to top Via using source IP:port.

Used by edge proxies / P-CSCFs for NAT traversal on REGISTER.

fix_nated_contact

fix_nated_contact() -> None

Rewrite Contact URI host:port with source IP:port.

Used for NAT traversal — ensures replies route back through the actual transport address rather than the Contact address the UA advertised.

force_send_via

force_send_via(transport: str, target: str) -> None

Override Via header transport and target for outgoing message.

Parameters:

Name Type Description Default
transport str

Protocol ("udp", "tcp", "tls").

required
target str

Target address (e.g. "10.0.0.2:5060").

required

generate_icid

generate_icid() -> str

Generate a unique ICID (IMS Charging ID) for P-Charging-Vector.

Returns:

Type Description
str

UUID string suitable for the icid-value parameter.

source_ip_in

source_ip_in(cidr_list: list[str]) -> bool

Check if the source IP is within any of the given CIDR ranges.

Parameters:

Name Type Description Default
cidr_list list[str]

List of CIDR strings (e.g. ["10.0.0.0/8"]).

required

Returns:

Type Description
bool

True if source_ip falls within any range.

Example::

if request.source_ip_in(["10.0.0.0/8", "172.16.0.0/12"]):
    log.info("Trusted network")

from_gateway

from_gateway(group_name: str) -> bool

Check if the source IP is a member of a gateway group.

Returns True when this request's source IP is one of the resolved addresses of the gateway group group_name (configured under gateway.groups in siphon.yaml, or via gateway.add_group). This is siphon's equivalent of Kamailio ds_is_from_list() / OpenSIPS ds_is_in_list() — a routing-direction / trust predicate that replaces hardcoded source CIDRs.

The match is on IP only (the source port is ignored — gateways answer from varied ports) and against every resolved address in the group, so a hostname that round-robins across many IPs (e.g. Teams' sip/sip2/sip3.pstnhub.microsoft.com) matches on any of them.

Infallible: returns False (never raises) when the group does not exist, no gateway is configured, or the source IP does not parse.

Security: on connection-oriented transports (TCP/TLS/WS/WSS) the source IP is handshake-verified and trustworthy as an authorization signal; on UDP it is spoofable, so from_gateway there is a best-effort direction hint, not an auth gate.

Parameters:

Name Type Description Default
group_name str

Name of the gateway group to test membership against.

required

Returns:

Type Description
bool

True if the source IP belongs to the group.

Example::

@proxy.on_request("INVITE")
def route(request):
    if request.from_gateway("teams"):
        # Inbound from Microsoft Teams — trust and forward to the PBX.
        request.relay("sip:pbx.internal:5060")
    else:
        request.reply(403, "Forbidden")