Reply¶

The Reply object wraps a SIP response. It reaches your script in a @proxy.on_reply handler (every response on a relayed transaction) and in a @proxy.on_failure handler (the best error after all branches failed).

from siphon import proxy

@proxy.on_reply
def observe(request, reply):
    if reply.status_code == 200 and request.method == "INVITE":
        reply.set_header("X-Answered-By", "siphon")
    reply.relay()

A SIP response message.

This object is passed as the second argument to reply handlers:

@proxy.on_reply — all responses
@proxy.on_failure — aggregated failure response
@proxy.on_register_reply — REGISTER responses

Example::

@proxy.on_reply
async def reply_route(request, reply):
    if reply.status_code == 200 and reply.has_body("application/sdp"):
        await rtpengine.answer(reply)
    reply.relay()

status_code `property` ¶

status_code: int

SIP status code (e.g. 200, 404, 503).

reason `property` ¶

reason: str

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

from_uri `property` ¶

from_uri: Optional[SipUri]

From header URI.

to_uri `property` ¶

to_uri: Optional[SipUri]

To header URI.

call_id `property` ¶

call_id: Optional[str]

Call-ID header value.

body `property` ¶

body: Optional[bytes]

Response body (e.g. SDP), or None.

content_type `property` ¶

content_type: Optional[str]

Content-Type header value.

source_ip `property` ¶

source_ip: Optional[str]

Source IP of the entity that sent this response, or None.

Populated on @proxy.on_reply, @proxy.on_failure per-relay callbacks, and the B2BUA @b2bua.on_answer / @b2bua.on_early_media replies (the B-leg peer that answered). None on a fork-aggregated @proxy.on_failure reply, where the "best" error is selected across branches and no single source applies. Reply-side counterpart of request.source_ip.

source_port `property` ¶

source_port: Optional[int]

Source port of the entity that sent this response, or None (see :attr:source_ip for when it is populated).

actions `property` ¶

actions: list[Action]

All actions recorded (test-only).

last_action `property` ¶

last_action: Optional[Action]

Most recent action, or None.

from_gateway ¶

from_gateway(group_name: str) -> bool

Check if this response's source IP is a member of a gateway group.

The reply-side counterpart of request.from_gateway / call.from_gateway — returns True when the source IP of the entity that sent this response is one of the resolved addresses of the gateway group group_name (configured under gateway.groups in siphon.yaml, or via gateway.add_group). Use it on a response to decide which trunk actually answered — siphon's answer to Kamailio ds_is_from_list() / OpenSIPS ds_is_in_list() on the reply path.

The match is on IP only (source port ignored) against every resolved address in the group, so a hostname that round-robins across many IPs matches on any of them.

Infallible: returns False (never raises) when the group does not exist, no gateway is configured, the response source is unknown (see :attr:source_ip), 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 response source IP belongs to the group.

Example::

@b2bua.on_answer
async def on_answer(call, reply):
    if reply.from_gateway("carriers"):
        log.info("answered by the carrier trunk")

get_header ¶

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

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

header ¶

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

Alias for :meth:get_header.

set_header ¶

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

Set (replace) a header value.

remove_header ¶

remove_header(name: str) -> None

Remove a header entirely.

has_header ¶

has_header(name: str) -> bool

Check if a header exists (case-insensitive).

has_body ¶

has_body(content_type: str) -> bool

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

Parameters:

Name	Type	Description	Default
`content_type`	`str`	MIME type (e.g. `"application/sdp"`).	required

take_av ¶

take_av()

Extract IMS-AKA CK/IK from auth headers and strip ck=/ik=.

Scans WWW-Authenticate, Proxy-Authenticate and Authentication-Info (in that order). Returns a :class:MockAuthVectorHandle only when both ck and ik parsed cleanly; otherwise leaves the headers untouched and returns None.

Idempotent: after stripping, a second call returns None because no header still carries ck/ik.

relay ¶

relay() -> None

Forward the reply upstream to the UAC.

Example::

@proxy.on_reply
def handle_reply(request, reply):
    reply.relay()

forward ¶

forward() -> None

Alias for :meth:relay.

reject ¶

reject(code: int, reason: Optional[str] = None) -> bool

Reject an in-progress proxied INVITE from @proxy.on_reply.

The proxy-side equivalent of the B2BUA's call.reject() — needed because media authorization (sbi.create_session / diameter.rx_aar) runs at answer time, when the negotiated SDP is available, and a media-authorization failure must reject the leg rather than proceed medialess.

Behaviour depends on the stage of the response this handler is running for:

Provisional (1xx) — no final answer yet: records the reject and returns True. siphon then sends code reason upstream to the UAC and CANCELs the pending downstream branch(es). This is the clean path — typically a reliable 183 Session Progress in the VoLTE preconditions / early-media flow, where the SDP answer rides the provisional.
Final (>= 200) — UAS already answered: a proxy cannot retract a 2xx, so this is a no-op and returns False. Branch on the return value: log the failed authorization and :meth:relay to let the answer through (best-effort, no dedicated bearer).

Takes precedence over :meth:relay / :meth:forward when it returns True.

Parameters:

Name	Type	Description	Default
`code`	`int`	SIP final-response code in the 400–699 range (e.g. `503`).	required
`reason`	`Optional[str]`	optional reason phrase; a sensible default is derived from `code` when omitted.	`None`

Returns:

Type	Description
`bool`	`True` if the reject was accepted (provisional stage — siphon will
`bool`	send the error + CANCEL); `False` if it could not be applied (the
`bool`	UAS already sent a final response).

Raises:

Type	Description
`ValueError`	if `code` is outside 400–699.

Example::

@proxy.on_reply
async def on_reply(request, reply):
    if request.method == "INVITE" and reply.has_body("application/sdp"):
        if not await authorize_media(request, reply):
            if reply.reject(503, "Media Authorization Failed"):
                return            # 503 + CANCEL sent by siphon
            log.warn("could not reject answered call; proceeding best-effort")
    reply.relay()

Reply¶

status_code property ¶

reason property ¶

from_uri property ¶

to_uri property ¶

call_id property ¶

body property ¶

content_type property ¶

source_ip property ¶

source_port property ¶

actions property ¶

last_action property ¶

from_gateway ¶

get_header ¶

header ¶

set_header ¶

remove_header ¶

has_header ¶

has_body ¶

take_av ¶

relay ¶

forward ¶

reject ¶

status_code `property` ¶

reason `property` ¶

from_uri `property` ¶

to_uri `property` ¶

call_id `property` ¶

body `property` ¶

content_type `property` ¶

source_ip `property` ¶

source_port `property` ¶

actions `property` ¶

last_action `property` ¶