IMS control¶

The namespaces that make SIPhon an IMS core: iFC evaluation (isc), 5G SBI / N5 policy authorization and Nbsf discovery (sbi), SIP presence (presence), lawful intercept (li), and the Session Recording Server hooks (srs).

`isc` namespace¶

Initial Filter Criteria evaluation (3GPP TS 29.228 / IMS Service Control).

A service profile may name shared iFC sets (<SharedIFCSetID>, TS 29.228 Annex B.2) that the S-CSCF holds locally, configured under isc::

isc:
  shared_ifc_sets:
    - id: 1
      file: /etc/siphon/sifc/1.xml   # <InitialFilterCriteria> elements

They are merged with the profile's own iFCs by priority at evaluation, and with one configured Cx SAR advertises the SiFC feature so the HSS sends the ids (TS 29.229 §7.1.1). A diverted request takes the ORIGINATING_CDIV iFCs, and SessionDescription trigger points read the SDP passed as body:

matches = isc.evaluate(
    served_user, request.method, str(request.ruri), headers,
    session_case="originating_cdiv",
    body=request.body if request.get_header("Content-Type") == "application/sdp" else None,
)

Mock ISC namespace — Initial Filter Criteria evaluation for testing.

Store per-user iFC profiles and evaluate them against requests.

Example::

from siphon_sdk import mock_module
mock_module.install()

from siphon import isc

# Store a profile (in mock, stores raw XML string)
count = isc.store_profile("sip:alice@example.com", ifc_xml)

# Evaluate — returns pre-configured matches
matches = isc.evaluate("sip:alice@example.com", "INVITE",
                       "sip:bob@example.com", [], "originating")

store_profile ¶

store_profile(aor: str, ifc_xml: str) -> int

Parse and store an iFC XML profile for an AoR.

A service profile may name shared iFC sets (<SharedIFCSetID>, TS 29.228 Annex B.2 / Annex E) instead of, or next to, its own iFCs; each is resolved against isc.shared_ifc_sets in siphon.yaml, and an id with no local set is logged at warn with the AoR.

In the mock, the XML is stored as-is (no actual parsing) apart from its SharedIFCSetIDs, which shared_set_ids() returns. Use set_eval_results() to configure what evaluate() returns.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address of Record.	required
`ifc_xml`	`str`	Raw iFC XML string.	required

Returns:

Type	Description
`int`	The number of iFC rules the profile applies: its own plus those
`int`	of the shared sets it names that the S-CSCF has (always 1 in the
`int`	mock).

Raises:

Type	Description
`ValueError`	If the XML cannot be parsed, or holds a `<SessionCase>` that is not 0 to 4 or a `<SharedIFCSetID>` that is not an integer of at least 0.

shared_set_ids ¶

shared_set_ids(aor: str) -> list[int]

The SharedIFCSetIDs the stored profile of an AoR names, for logging.

Returns:

Type	Description
`list[int]`	The ids in profile order; empty when it names none or no profile
`list[int]`	is stored.

remove_profile ¶

remove_profile(aor: str) -> bool

Remove a stored profile.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address of Record.	required

Returns:

Type	Description
`bool`	`True` if a profile was removed.

has_profile ¶

has_profile(aor: str) -> bool

Check if a profile is stored for an AoR.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address of Record.	required

Returns:

Type	Description
`bool`	`True` if a profile exists.

evaluate ¶

evaluate(
    aor: str,
    method: str,
    ruri: str,
    headers: "list[tuple[str, str]]",
    session_case: str = "originating",
    start_after_priority: Optional[int] = None,
    body: "Optional[str | bytes]" = None,
) -> list[dict]

Evaluate iFCs for a request.

The stored profile's own iFCs and those of the shared iFC sets it names are merged by priority.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address of Record.	required
`method`	`str`	SIP method (e.g. `"INVITE"`).	required
`ruri`	`str`	Request-URI string.	required
`headers`	`'list[tuple[str, str]]'`	List of (name, value) tuples.	required
`session_case`	`str`	`"originating"`, `"terminating"`, `"originating_unregistered"`, `"terminating_unregistered"` or `"originating_cdiv"` (TS 29.228 Annex E value 4: the originating services after retargeting, TS 24.229 §5.4.3.3 NOTE 4); short forms `orig`, `term`, `orig_unreg`, `term_unreg`, `orig_cdiv`.	`'originating'`
`start_after_priority`	`Optional[int]`	Only iFCs with a higher priority, to resume after the AS the request came back from.	`None`
`body`	`'Optional[str \| bytes]'`	The request's SDP (`str` or `bytes`), for `<SessionDescription>` trigger points (TS 29.228 Annex F: `Line` matched against each field's type, `Content` against its value). `None` for a request without SDP, which such a trigger point does not match. For a multipart body, pass the `application/sdp` part.	`None`

Returns:

Name	Type	Description
	`list[dict]`	List of dicts with keys: `server_name`, `default_handling`,
	`list[dict]`	`service_info`, `priority`, `include_register_request`
	`list[dict]`	(bool) and `include_register_response` (bool).
	`list[dict]`	The two flags are the iFC's `ApplicationServer/Extension`
	`list[dict]`	elements of TS 29.228 (empty elements: present means set). When
	`list[dict]`	true, the S-CSCF includes the user's REGISTER, and the 200 OK it
	`list[dict]`	answered with, in the third-party REGISTER to that AS
	`list[dict]`	(TS 24.229 §5.4.1.7).
`Example`	`S - CSCF`	: for match in isc.evaluate(aor, "REGISTER", aor, []): wants_request = match["include_register_request"] wants_response = match["include_register_response"] if wants_request or wants_response: log.info(f"{match['server_name']} wants the REGISTER " f"(request={wants_request}, response={wants_response})")

Raises:

Type	Description
`ValueError`	`session_case` is not one of the above.

In the mock, returns what set_eval_results configured (an empty list otherwise), without priorities at or below start_after_priority; every call is recorded on evaluations.

profile_count ¶

profile_count() -> int

Number of stored per-user iFC profiles.

set_eval_results ¶

set_eval_results(aor: str, results: list[dict]) -> None

Configure what evaluate() returns for a given AoR.

Parameters:

Name	Type	Description	Default
`aor`	`str`	Address of Record.	required
`results`	`list[dict]`	List of dicts, each with keys `server_name`, `default_handling`, `service_info`, `priority` and optionally `include_register_request` / `include_register_response` (both default `False`).	required

Example::

isc.set_eval_results("sip:alice@example.com", [
    {"server_name": "sip:as1@example.com", "default_handling": 0,
     "service_info": None, "priority": 0},
])

clear ¶

clear() -> None

Reset all stored profiles and evaluation results.

`sbi` namespace¶

5G Service-Based Interface — N5/Npcf policy authorization plus Nbsf_Management PCF discovery.

PCF callbacks¶

The PCF calls back over HTTP on the address in sbi.notif_listen. Advertise http://<notif_listen>/sbi/events as notif_uri; the PCF appends the TS 29.514 suffix for each callback, so one listener serves both.

notifUri is a required attribute of the create (TS 29.514), so sbi.create_session needs a callback URI on every call, with or without events. Set it once in the configuration and a script can leave the argument out; notif_uri= on a call overrides it. With neither, the call raises ValueError and nothing is sent.

sbi:
  notif_listen: "0.0.0.0:8080"
  notif_uri: "http://192.0.2.10:8080/sbi/events"   # how the PCF reaches notif_listen

Route	Body (verbatim dict)	Hook
`POST /sbi/events/notify`	`EventsNotification`	`@sbi.on_event`
`POST /sbi/events/terminate`	`TerminationInfo`	`@sbi.on_terminate`

Both answer 204 once the handlers ran (a handler that raises is logged and still acknowledged), 400 for a body that is not JSON, and 503 when siphon's Python executor could not take the job, so the PCF knows the callback was not handled. Any other path is 404.

A termination can arrive for any app session created with notif_uri. Events only arrive for what the session subscribed to, with create_session(events=[...], notif_uri=...) or update_session(events=[...]).

The bare POST /sbi/events, which reached @sbi.on_event for a PCF that posted to the advertised URI without a suffix, was deprecated in 1.9.0 and removed in 1.10.0. It now answers 404 like any other unknown path. A PCF that appends the TS 29.514 suffix — which is what the spec defines — is unaffected.

Mock SBI namespace for testing scripts that use from siphon import sbi.

Provides mock N5/Npcf policy authorization methods plus Nbsf_Management discovery (discover_pcf_binding).

Example::

import asyncio
from siphon_sdk import mock_module
mock_module.install()

from siphon import sbi
# Awaitable, as in siphon, so a synchronous test drives it through
# asyncio.run; an async test just awaits it.
result = asyncio.run(sbi.create_session(sip_call_id="call-1", ue_ipv4="10.0.0.1"))
assert result["authorized"] is True

set_notif_uri ¶

set_notif_uri(notif_uri: Optional[str]) -> None

Test helper: act as a node with sbi.notif_uri configured (or, with None, without it).

create_session ¶

create_session(
    af_app_id: str = "IMS Services",
    sip_call_id: Optional[str] = None,
    supi: Optional[str] = None,
    ue_ipv4: Optional[str] = None,
    ue_ipv6: Optional[str] = None,
    dnn: Optional[str] = None,
    notif_uri: Optional[str] = None,
    media_components: Optional[list] = None,
    pcf_uri: Optional[str] = None,
    events: Optional[list[str]] = None,
) -> Optional[dict]

Create an N5 app session for QoS policy authorization.

Parameters:

Name	Type	Description	Default
`af_app_id`	`str`	AF-Application identifier (default `"IMS Services"`).	`'IMS Services'`
`sip_call_id`	`Optional[str]`	SIP Call-ID of the call, for correlation in siphon's own log. It is not sent to the PCF: TS 29.514 has no such attribute.	`None`
`supi`	`Optional[str]`	Subscription Permanent Identifier.	`None`
`ue_ipv4`	`Optional[str]`	UE IPv4 address.	`None`
`ue_ipv6`	`Optional[str]`	UE IPv6 address.	`None`
`dnn`	`Optional[str]`	Data Network Name.	`None`
`notif_uri`	`Optional[str]`	base URI of siphon's PCF callback listener, `http://<sbi.notif_listen>/sbi/events`. Defaults to `sbi.notif_uri` from the configuration; one of the two must be set, since TS 29.514 requires it. Sent as `notifUri`, where the PCF posts a termination (`/terminate`, to `@sbi.on_terminate`), and with `events` also as the subscription's `notifUri`, where it posts events (`/notify`, to `@sbi.on_event`).	`None`
`media_components`	`Optional[list]`	list of media-component dicts (same shape as `diameter.rx_aar`'s `media_components`).	`None`
`pcf_uri`	`Optional[str]`	per-call N5 target — address this session at the given PCF base URL (e.g. a BSF-discovered `pcf_uri`) instead of the configured `npcf_url`. `None` ⇒ configured PCF.	`None`
`events`	`Optional[list[str]]`	PCF events to subscribe to, by TS 29.514 `AfEvent` name (e.g. `"FAILED_RESOURCES_ALLOCATION"`, `"SUCCESSFUL_RESOURCES_ALLOCATION"`, `"QOS_NOTIF"`). Each is sent with `notifMethod` `"EVENT_DETECTION"`. Names are passed through unchecked, so events newer than siphon work. Requires `notif_uri`. `None` (default) subscribes to nothing, and the PCF sends no events.	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `app_session_id`, `authorized` and `app_session_uri`
`Optional[dict]`	(the absolute resource URI — persist it and hand it back to
`Optional[dict]`	`update_session` / `delete_session` for replica-independent
`Optional[dict]`	teardown), or `None`.

Raises:

Type	Description
`ValueError`	neither `notif_uri` nor `sbi.notif_uri` is set (`notifUri` is required on a create), or `events` is empty.
`TypeError`	`events` is not a list of strings.

Example::

result = await sbi.create_session(
    sip_call_id=request.call_id,
    ue_ipv4=request.source_ip,
    notif_uri="http://192.0.2.10:8080/sbi/events",
    events=["FAILED_RESOURCES_ALLOCATION"],
)

delete_session `async` ¶

delete_session(session_id: str) -> bool

Delete an N5 app session.

True means the session no longer exists on the PCF: siphon deleted it (the PCF answered 2xx), or it was already gone (the PCF answered 404). The second is the usual answer to the delete an @sbi.on_terminate handler makes with termination["resUri"]. In both cases siphon stops tracking the session, so siphon_sbi_npcf_app_sessions_active comes back down.

False means the delete failed (a transport error or any other non-2xx answer); siphon keeps tracking the session.

In this mock an unknown session is treated as the PCF's 404 and returns True; use set_delete_failure to exercise False.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The app session id from `create_session()` or the absolute `app_session_uri` (replica-independent teardown).	required

Returns:

Type	Description
`bool`	`True` when the session is gone (deleted or already removed),
`bool`	`False` when the delete failed.

Example::

@sbi.on_terminate
async def handle_termination(termination):
    if not await sbi.delete_session(termination["resUri"]):
        log.warn("app session delete failed; PCF still holds it")

update_session ¶

update_session(
    session_id: str,
    media_components: Optional[list] = None,
    events: Optional[list[str]] = None,
    notif_uri: Optional[str] = None,
) -> Optional[dict]

Update an N5 app session (media renegotiation, event subscription).

The modify is a JSON merge patch, so only what you pass is sent.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The app session id to update, or the absolute `app_session_uri` from `create_session`.	required
`media_components`	`Optional[list]`	list of media-component dicts (same shape as `create_session`).	`None`
`events`	`Optional[list[str]]`	replace the subscribed PCF events with these (same names as `create_session`). `None` (default) sends no subscription and leaves the one the PCF holds untouched. Removing the subscription is not possible from here.	`None`
`notif_uri`	`Optional[str]`	a new callback base for the subscription (`http://<sbi.notif_listen>/sbi/events`). Only sent with `events`; `None` keeps the one the PCF holds.	`None`

Returns:

Type	Description
`Optional[dict]`	Dict with `app_session_id` and `authorized`, or `None`.

Raises:

Type	Description
`ValueError`	`events` is empty, or `notif_uri` is given without `events`.
`TypeError`	`events` is not a list of strings.

discover_pcf_binding ¶

discover_pcf_binding(
    ue_ipv4: Optional[str] = None,
    ue_ipv6: Optional[str] = None,
) -> Awaitable[Optional[dict]]

Nbsf_Management discovery — look up the PCF binding for a UE IP.

Returns a binding dict (5G; configure via set_binding), None when the BSF has no binding (it answers 204, or 404 on some servers; 4G), or raises sbi.BsfError when configured unhealthy via set_bsf_error.

Exactly one of ue_ipv4 / ue_ipv6 must be supplied.

Parameters:

Name	Type	Description	Default
`ue_ipv4`	`Optional[str]`	UE IPv4 address (the IPsec SA peer).	`None`
`ue_ipv6`	`Optional[str]`	UE IPv6 address. siphon sends it as the `ipv6Prefix` query parameter with `/128` appended, as TS 29.521 requires; a value that already carries a prefix length is sent as written.	`None`

Returns:

Type	Description
`Awaitable[Optional[dict]]`	The binding dict (incl. a ready-to-use `pcf_uri`) or `None`.

on_event `staticmethod` ¶

on_event(fn: Any) -> Any

Register a handler for incoming PCF event notifications (N5).

The handler receives the PCF's EventsNotification document (TS 29.514 §5.6.2.6) verbatim as a dict — every field is preserved, so the keys are the exact 3GPP wire names. Use evSubsUri to correlate the event with the app-session you created, and evNotifs for the per-event list. Each entry's flows carries medCompN + fNums (not flow descriptions).

siphon serves the callback on sbi.notif_listen at POST /sbi/events/notify: advertise http://<notif_listen>/sbi/events as notif_uri and the PCF appends /notify. The PCF only sends events you subscribed to with create_session(events=[...]) or update_session(events=[...]).

Sync and async handlers both work. A handler that raises is logged and the PCF still gets 204: a retry would hit the same bug. When siphon's Python executor is saturated and the handler cannot run, the PCF gets 503 so it knows the notification was not taken.

POST /sbi/events (no suffix) also reaches this hook, for a PCF that posts to the advertised URI as-is. It is deprecated and goes away in the next minor release.

Parameters:

Name	Type	Description	Default
`fn`	`Any`	`def handler(event: dict) -> None` or the `async def` equivalent. `event` is the `EventsNotification` dict.	required

Returns:

Type	Description
`Any`	`fn` unchanged, so it stays callable from tests.

Example::

@sbi.on_event
def handle_pcf_event(event):
    session_events_uri = event.get("evSubsUri")
    for notif in event.get("evNotifs", []):
        if notif["event"] == "FAILED_RESOURCES_ALLOCATION":
            log.warn(f"PCF could not allocate resources: {session_events_uri}")

on_terminate `staticmethod` ¶

on_terminate(fn: Any) -> Any

Register a handler for PCF-initiated app-session termination (N5).

The PCF sends this when the PDU session behind an app session is released (the N5 counterpart of a Diameter Rx ASR). The handler receives the PCF's TerminationInfo document (TS 29.514) verbatim as a dict:

termCause: why, e.g. "PDU_SESSION_TERMINATION" or "ALL_SDF_DEACTIVATION".
resUri: the app-session resource URI, the same value create_session returned as app_session_uri. Use it to find the call the session belonged to.

TS 29.514 has the AF acknowledge the termination and then delete the app session, so a handler normally releases its per-call state and calls sbi.delete_session(termination["resUri"]). That also drops siphon's own tracking of the session.

This is a separate hook from on_event on purpose: an event leaves the session in place, a termination means it is ending. siphon serves it on sbi.notif_listen at POST /sbi/events/terminate; the PCF appends /terminate to the notif_uri given to create_session, so the same http://<notif_listen>/sbi/events base serves both callbacks. Answers follow on_event: 204 once the handlers ran (a raising handler is logged), 503 when the executor could not run them.

Parameters:

Name	Type	Description	Default
`fn`	`Any`	`def handler(termination: dict) -> None` or the `async def` equivalent.	required

Returns:

Type	Description
`Any`	`fn` unchanged, so it stays callable from tests.

Example::

@sbi.on_terminate
async def handle_termination(termination):
    log.warn(f"PCF ended app session: {termination['termCause']}")
    await sbi.delete_session(termination["resUri"])

set_authorized ¶

set_authorized(authorized: bool) -> None

Configure whether create_session returns authorized (test helper).

Parameters:

Name	Type	Description	Default
`authorized`	`bool`	Whether sessions should be authorized.	required

set_binding ¶

set_binding(binding: Optional[dict]) -> None

Configure what discover_pcf_binding returns (test helper).

Parameters:

Name	Type	Description	Default
`binding`	`Optional[dict]`	a binding dict (5G case) or `None` (no binding: the 4G case).	required

set_bsf_error ¶

set_bsf_error(raise_error: bool) -> None

Configure discover_pcf_binding to raise BsfError (test helper).

Parameters:

Name	Type	Description	Default
`raise_error`	`bool`	when True, `discover_pcf_binding` raises `BsfError`.	required

set_delete_failure ¶

set_delete_failure(fail: bool) -> None

Make delete_session fail (test helper).

Stands in for a PCF that answers the delete with an error other than 404, or cannot be reached: delete_session returns False and the session stays tracked.

Parameters:

Name	Type	Description	Default
`fail`	`bool`	when True, `delete_session` returns `False`.	required

clear ¶

clear() -> None

Reset all mock sessions and failure switches (test helper).

`BsfError`¶

Raised by sbi.discover_pcf_binding(...) when the BSF is unhealthy. A BSF that has no binding for the address is healthy: it answers 204 (TS 29.521 §4.2.4.2; some servers answer 404) and the lookup returns None.

Bases: RuntimeError

Raised by sbi.discover_pcf_binding() when the BSF is unhealthy (5xx / timeout / transport / malformed body).

No binding for the UE IP is not a BsfError: the BSF answers 204 (TS 29.521), or 404 on some servers, and the lookup returns None (the 4G UE case). Mirrors the Rust sbi.BsfError exception.

`presence` namespace¶

SIP presence document publish/lookup and subscription tracking (RFC 3856 / 6665).

Mock presence namespace — SIP presence publish/subscribe for testing.

Manages presence documents and subscriptions in-memory.

Example::

from siphon_sdk import mock_module
mock_module.install()

from siphon import presence

etag = presence.publish("sip:alice@example.com", "<presence/>", expires=3600)
doc = presence.lookup("sip:alice@example.com")
assert doc == "<presence/>"

sub_id = presence.subscribe("sip:bob@example.com", "sip:alice@example.com")
watchers = presence.subscribers("sip:alice@example.com")
assert len(watchers) == 1

Test helper::

from siphon_sdk.mock_module import get_presence
p = get_presence()
p.clear()

Answer the next NOTIFYs as a subscriber would, to test what a script does once a subscription ends::

p.set_notify_response(481)   # the subscriber lost the dialog
p.set_notify_response(None)  # no answer: Timer F expires

notifications `property` ¶

notifications: list

List of NOTIFY messages sent (for test assertions).

publish ¶

publish(
    entity: str, pidf_xml: str, expires: int = 3600
) -> str

Publish a presence document for a presentity.

Parameters:

Name	Type	Description	Default
`entity`	`str`	Presentity URI (e.g. `"sip:alice@example.com"`).	required
`pidf_xml`	`str`	PIDF XML body string.	required
`expires`	`int`	Document expiry in seconds (default: 3600).	`3600`

Returns:

Type	Description
`str`	An etag string assigned to the published document.

Example::

etag = presence.publish("sip:alice@example.com",
                         "<presence><tuple><status><basic>open</basic></status></tuple></presence>")

lookup ¶

lookup(entity: str) -> Optional[str]

Look up the current presence document for a URI.

Parameters:

Name	Type	Description	Default
`entity`	`str`	Presentity URI to look up.	required

Returns:

Type	Description
`Optional[str]`	PIDF XML string, or `None` if not found.

subscribe ¶

subscribe(
    subscriber: str,
    resource: str,
    event: str = "presence",
    expires: int = 3600,
) -> str

Subscribe to presence for a resource.

Creates a new subscription and returns its ID.

Parameters:

Name	Type	Description	Default
`subscriber`	`str`	Watcher URI (e.g. `"sip:bob@example.com"`).	required
`resource`	`str`	Presentity URI to watch.	required
`event`	`str`	Event package name (default: `"presence"`).	`'presence'`
`expires`	`int`	Subscription duration in seconds (default: 3600).	`3600`

Returns:

Type	Description
`str`	Subscription ID string.

subscribe_dialog ¶

subscribe_dialog(
    subscriber: str,
    resource: str,
    event: str = "reg",
    expires: int = 3600,
    call_id: str = "",
    from_tag: str = "",
    to_tag: str = "",
    route_set: Optional[list] = None,
    local_uri: Optional[str] = None,
    remote_uri: Optional[str] = None,
    local_cseq: int = 0,
    version: int = 0,
) -> str

Create a subscription with dialog state for in-dialog NOTIFY.

Parameters:

Name	Type	Description	Default
`subscriber`	`str`	Watcher Contact URI from the SUBSCRIBE — the dialog's remote target. Used for the NOTIFY's Request-URI and to resolve where to send it; it is not what goes in `To` (see `remote_uri`).	required
`resource`	`str`	Presentity URI being watched.	required
`event`	`str`	Event package name.	`'reg'`
`expires`	`int`	Subscription duration in seconds.	`3600`
`call_id`	`str`	Call-ID from the SUBSCRIBE dialog.	`''`
`from_tag`	`str`	From-tag from the SUBSCRIBE.	`''`
`to_tag`	`str`	To-tag from the SUBSCRIBE.	`''`
`route_set`	`Optional[list]`	Route headers from Record-Route. Each entry is kept as one `name-addr` whatever form it is written in (a bare URI, a bracketed one, a full `name-addr`); one that is not a SIP or SIPS URI raises `ValueError`.	`None`
`local_uri`	`Optional[str]`	The SUBSCRIBE's To URI — the dialog's local URI, which RFC 3261 §12.2.1.1 requires in the `From` of every in-dialog NOTIFY. Pass `str(request.to_uri)`.	`None`
`remote_uri`	`Optional[str]`	The SUBSCRIBE's From URI — the dialog's remote URI, required in the `To`. Pass `str(request.from_uri)`.	`None`
`local_cseq`	`int`	The CSeq number of the dialog's last NOTIFY; the next carries one more. `0` for a new dialog; a dialog rebuilt from stored state (IMS restoration) starts high enough for the subscriber to accept it. Kept as `local_cseq` in the record.	`0`
`version`	`int`	The version the next event-state document carries (RFC 3680 §5.2), handed out by :meth:`next_version`. `0` for a new subscription. IMS restoration stores no version, so a rebuilt dialog passes one above any it could have sent, as for `local_cseq`: a subscriber that sees a version more than one higher takes it and SHOULD refresh for full state.	`0`

Both URI arguments default to resource, which is correct for any package where the subscriber watches an AoR directly (the reg event package). Supply them for a watcher subscribed to somebody else's resource, where the remote URI is the watcher's own AoR.

Returns:

Type	Description
`str`	Subscription ID string.

unsubscribe ¶

unsubscribe(subscription_id: str) -> bool

Unsubscribe by subscription ID.

Parameters:

Name	Type	Description	Default
`subscription_id`	`str`	The subscription ID returned by :meth:`subscribe`.	required

Returns:

Type	Description
`bool`	`True` if the subscription was found and removed.

refresh ¶

refresh(subscription_id: str, expires: int) -> bool

Refresh a subscription's expiry (RFC 6665 §4.4.1 re-SUBSCRIBE).

Resets the subscription timer to expires seconds, keeping the dialog. Pair with :meth:find_by_dialog to resolve the id from an in-dialog SUBSCRIBE before refreshing.

Parameters:

Name	Type	Description	Default
`subscription_id`	`str`	The subscription ID (from `subscribe*` or :meth:`find_by_dialog`).	required
`expires`	`int`	New subscription duration in seconds.	required

Returns:

Type	Description
`bool`	`True` if the subscription was found and refreshed.

find_by_dialog ¶

find_by_dialog(
    call_id: str, from_tag: str
) -> Optional[str]

Resolve a subscription id from its dialog (Call-ID, From-tag).

An in-dialog SUBSCRIBE (a refresh, or an un-SUBSCRIBE with Expires: 0) carries the dialog's Call-ID and the subscriber's From-tag but not the original subscription id. This maps that pair back so a notifier (e.g. an S-CSCF handling reg-event) can :meth:refresh or :meth:unsubscribe the right dialog. Only subscriptions created with :meth:subscribe_dialog (which store dialog state) are findable.

Parameters:

Name	Type	Description	Default
`call_id`	`str`	Call-ID of the in-dialog SUBSCRIBE.	required
`from_tag`	`str`	From-tag of the in-dialog SUBSCRIBE (subscriber's tag).	required

Returns:

Type	Description
`Optional[str]`	The subscription ID string, or `None` if no dialog matches.

next_version ¶

next_version(subscription_id: str) -> Optional[int]

The version for the next event-state document on a subscription.

RFC 3680 §5.2: "Versions start at 0, and increment by one for each new document sent to a subscriber. Versions are scoped within a subscription." The store owns one count per subscription; each call returns it and advances it, so take a version only for a document that is sent.

Parameters:

Name	Type	Description	Default
`subscription_id`	`str`	The subscription ID.	required

Returns:

Type	Description
`Optional[int]`	The version (0 for the first document), or `None` for an unknown
`Optional[int]`	subscription. A removed subscription's count goes with it.

Example::

body = registrar.reginfo_xml(aor, state="partial",
                             version=presence.next_version(sub_id))
await presence.notify(sub_id, body=body,
                      content_type="application/reginfo+xml")

subscribers ¶

subscribers(resource: str) -> list[dict]

List subscribers (watchers) for a resource.

Parameters:

Name	Type	Description	Default
`resource`	`str`	Presentity URI to query.	required

Returns:

Type	Description
`list[dict]`	List of dicts with keys: `id`, `subscriber`, `event`.

subscription_count ¶

subscription_count() -> int

Get the total number of subscriptions.

document_count ¶

document_count() -> int

Get the total number of entities with published documents.

notify ¶

notify(
    subscription_id: str,
    body: Optional[str] = None,
    content_type: Optional[str] = None,
    subscription_state: str = "active",
    wait_for_response: bool = False,
    timeout_ms: int = 2000,
) -> Optional[int]

Send an in-dialog NOTIFY for a subscription. Awaitable.

In the mock, this records the notification for test assertions and answers it with the status set by :meth:set_notify_response (200 unless set).

When subscription_state indicates a terminated subscription (RFC 6665 §4.1.3 — bare "terminated" or "terminated;reason=...") the subscription is also removed from the store, mirroring the production auto-GC behavior.

The NOTIFY's final response is always acted on, whether or not the script waits for it. A subscriber that lost the dialog answers 481; that, any other of the responses RFC 6665 §4.2.2 names (404, 405, 410, 416, 480-485, 489, 501, 604), a 408, or no final response before Timer F removes the subscription, in the background, so the next :meth:subscribers no longer lists it. It is counted in siphon_presence_subscriptions_removed_total{reason} (notify_rejected / notify_timeout). A refresh that arrives while the NOTIFY is out keeps the subscription: the subscriber still holds the dialog. Any other failure (a 5xx, say) only failed the transaction and keeps it.

Parameters:

Name	Type	Description	Default
`subscription_id`	`str`	The subscription ID from `subscribe_dialog()`.	required
`body`	`Optional[str]`	Optional body string (reginfo XML, PIDF XML, etc.).	`None`
`content_type`	`Optional[str]`	Content-Type of the body.	`None`
`subscription_state`	`str`	Subscription-State header value (default `"active"`).	`'active'`
`wait_for_response`	`bool`	`False` (the default) resolves as soon as the NOTIFY is handed to the transport, as it always has; the removal above happens without the script. `True` resolves to the final status code instead. Do not wait on the initial NOTIFY inside the SUBSCRIBE handler that accepts the subscription: that NOTIFY is held until the handler's 200 OK is out (RFC 6665 §4.1.2.3), so the wait only ends at `timeout_ms`.	`False`
`timeout_ms`	`int`	How long `wait_for_response=True` waits before resolving to `None`.	`2000`

Returns:

Type	Description
`Optional[int]`	`None`; with `wait_for_response=True` the final status
`Optional[int]`	(`int`), or `None` when none arrived within `timeout_ms`.

Example::

status = await presence.notify(sub_id, body=xml,
                               content_type="application/reginfo+xml",
                               wait_for_response=True)
if status == 481:
    log.info("watcher gone, subscription removed")

set_notify_response ¶

set_notify_response(status: Optional[int]) -> None

Answer the NOTIFYs that follow with status (test helper).

None models a subscriber that never answers: Timer F expires.

terminate ¶

terminate(
    subscription_id: str,
    reason: Optional[str] = None,
    body: Optional[str] = None,
    content_type: Optional[str] = None,
) -> bool

Send a terminating NOTIFY and remove the subscription (RFC 6665 §4.4.1).

Sends an in-dialog NOTIFY with Subscription-State: terminated;reason=<reason>, then removes the subscription's dialog state from the store. Idempotent: a second call with the same subscription_id returns False.

Parameters:

Name	Type	Description	Default
`subscription_id`	`str`	The subscription ID from `subscribe_dialog()`.	required
`reason`	`Optional[str]`	Termination reason per RFC 6665 §4.2.2 — one of `"deactivated"`, `"probation"`, `"rejected"`, `"timeout"`, `"giveup"`, `"noresource"`, `"invariant"`. Defaults to `"noresource"`.	`None`
`body`	`Optional[str]`	Optional final body.	`None`
`content_type`	`Optional[str]`	Content-Type of the body.	`None`

Returns:

Type	Description
`bool`	`True` if the subscription existed and the NOTIFY was
`bool`	recorded; `False` if the `subscription_id` was unknown.

Example::

sub_id = presence.subscribe_dialog(...)
...
await presence.terminate(sub_id, reason="timeout")

parse_reginfo ¶

parse_reginfo(xml: str) -> dict

Parse an RFC 3680 application/reginfo+xml body for tests.

Mirrors the Rust presence.parse_reginfo shape — returns a dict {"version": int, "state": "full"|"partial", "registrations": [...]} so tests asserting against script logic can use the same dict layout the production binary returns.

clear ¶

clear() -> None

Reset all documents, subscriptions, and notifications (test helper).

`li` namespace¶

Lawful intercept (ETSI X1/X2/X3) and SIPREC recording triggers.

Mock li namespace — lawful intercept operations for testing.

.. note:: Interception is not triggered from a script. siphon matches every SIP message against the warrants the ADMF provisioned over ETSI X1, in the dispatcher, on every leg — a warrant applies whether or not a script calls anything here. This namespace is for visibility (is this call warranted?) and for operator-driven SIPREC recording, which is not a warrant.

``intercept()`` and ``stop_intercept()`` are kept so existing scripts
keep working, but they only **report** whether a warrant matched. If
they still triggered, a script that called them would produce duplicate
IRI records for one event.

Pre-configure targets for testing::

from siphon_sdk.mock_module import get_li
li = get_li()
li.add_target("sip:alice@example.com")

Then in your script::

from siphon import li
if li.is_target(request):
    log.info("this call is subject to a warrant")

Test assertions::

li = get_li()
assert li.is_target(request)
assert li.events == [("record", "call-1@example.com")]

is_enabled `property` ¶

is_enabled: bool

Whether the LI subsystem is enabled.

In the mock, returns True if _enabled is set and targets are configured.

task_count `property` ¶

task_count: int

How many intercept tasks the ADMF has provisioned over X1.

Read-only: warrants are provisioned by the ADMF, never by a script.

destination_count `property` ¶

destination_count: int

How many delivery destinations the ADMF has provisioned over X1.

events `property` ¶

events: list[tuple[str, str]]

List of (operation, target_or_call_id) tuples recorded.

Operations: "intercept", "record", "stop_intercept", "stop_recording".

targets `property` ¶

targets: list[str]

List of currently configured target URIs.

is_target ¶

is_target(request: Any) -> bool

Check if a request matches an active intercept target.

Matches From URI, To URI, or RURI against configured targets.

Parameters:

Name	Type	Description	Default
`request`	`Any`	The SIP request object.	required

Returns:

Type	Description
`bool`	`True` if the request matches any configured target.

intercept ¶

intercept(request: Any) -> bool

Report whether this request is being intercepted.

Parameters:

Name	Type	Description	Default
`request`	`Any`	The SIP request object.	required

Returns:

Type	Description
`bool`	`True` if a provisioned warrant matches.

.. note:: Retained for compatibility. This does not trigger interception — the dispatcher has already emitted the IRI record for any matching message before a script handler runs. Calling it is harmless and changes nothing.

record ¶

record(target: Any) -> bool

Start SIPREC recording for a request or call.

Accepts either a Request (proxy mode) or Call (B2BUA mode). In B2BUA mode, the dispatcher will start SIPREC recording on answer using the SRS URI from lawful_intercept.siprec.srs_uri config.

SIPREC is a recording feature, not lawful interception: it produces no X2 record and is not tied to a provisioned warrant.

Parameters:

Name	Type	Description	Default
`target`	`Any`	A `Request` or `Call` object.	required

Returns:

Type	Description
`bool`	`True` if recording was initiated.

Example::

@b2bua.on_invite
def on_invite(call):
    li.record(call)       # B2BUA mode
    call.dial("sip:bob@example.com")

@proxy.on_request("INVITE")
def on_invite(request):
    li.record(request)    # proxy mode
    request.relay()

stop_intercept ¶

stop_intercept(request: Any) -> bool

Report whether this request is being intercepted.

Parameters:

Name	Type	Description	Default
`request`	`Any`	The SIP request object.	required

Returns:

Type	Description
`bool`	`True` if a provisioned warrant matches.

.. note:: Retained for compatibility. Session teardown records are emitted by the dispatcher when the dialog ends; this does not emit one.

set_provisioned_counts ¶

set_provisioned_counts(
    tasks: int, destinations: int
) -> None

Test helper: set what task_count / destination_count report.

Not part of the siphon API — the real counts come from what the ADMF provisioned.

stop_recording ¶

stop_recording(target: Any) -> bool

Stop SIPREC recording for a request or call.

Accepts either a Request or Call object.

Parameters:

Name	Type	Description	Default
`target`	`Any`	A `Request` or `Call` object.	required

Returns:

Type	Description
`bool`	`True` if a stop event was emitted.

add_target ¶

add_target(uri: str) -> None

Add a target URI for intercept matching (test helper).

Parameters:

Name	Type	Description	Default
`uri`	`str`	SIP URI to match against (e.g. `"sip:alice@example.com"`).	required

clear ¶

clear() -> None

Reset targets, events, and enabled state (test helper).

`srs` namespace¶

Session Recording Server acceptance hooks (RFC 7866 SIPREC).

Mock srs namespace — Session Recording Server hooks for testing.

Pre-configure accept/reject behavior::

from siphon_sdk.mock_module import get_srs
srs = get_srs()
srs.accept_all = False          # reject all recordings

Register handlers as in production::

from siphon import srs

@srs.on_invite
async def on_recording(metadata):
    return True

@srs.on_session_end
async def on_recording_end(session):
    pass

Inspect events after test::

srs = get_srs()
assert len(srs.sessions) == 1

accept_all `property` `writable` ¶

accept_all: bool

Whether mock auto-accepts all recordings (default True).

sessions `property` ¶

sessions: list[dict[str, Any]]

List of completed recording sessions (for test assertions).

invite_events `property` ¶

invite_events: list[dict[str, Any]]

List of on_invite calls received (for test assertions).

on_invite ¶

on_invite(fn: Any) -> Any

Register handler for incoming SIPREC INVITE (recording request).

The handler receives (metadata,) where metadata is a :class:~siphon_sdk.srs.RecordingMetadata object.

Return True to accept the recording, False to reject (403).

Example::

@srs.on_invite
async def on_recording(metadata):
    log.info(f"Recording: {metadata.session_id}")
    return True

on_session_end ¶

on_session_end(fn: Any) -> Any

Register handler for recording session completion.

The handler receives (session,) where session is a :class:~siphon_sdk.srs.SrsSession object.

Example::

@srs.on_session_end
async def on_recording_end(session):
    log.info(f"Recording {session.session_id} done")

record_invite ¶

record_invite(
    session_id: str, participants: list[str] | None = None
) -> None

Test helper: simulate an inbound SIPREC INVITE event.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Recording session identifier.	required
`participants`	`list[str] \| None`	List of participant AoRs.	`None`

record_session_end ¶

record_session_end(
    session_id: str,
    recording_call_id: str = "",
    duration_secs: int = 0,
    recording_dir: str | None = None,
) -> None

Test helper: simulate a completed recording session.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	Recording session identifier.	required
`recording_call_id`	`str`	Call-ID of the SIPREC dialog.	`''`
`duration_secs`	`int`	Recording duration in seconds.	`0`
`recording_dir`	`str \| None`	Path where recordings were written.	`None`

clear ¶

clear() -> None

Reset all mock state (test helper).

`SrsSession`¶

A completed recording session.

Completed recording session info.

Passed to @srs.on_session_end handlers after the recording finishes (BYE from SRC or timeout).

Example::

@srs.on_session_end
async def on_recording_end(session):
    log.info(f"Recording {session.session_id} complete")
    log.info(f"Duration: {session.duration}s")
    if session.recording_dir:
        log.info(f"Files in: {session.recording_dir}")

session_id `property` ¶

session_id: str

SRS session identifier.

recording_call_id `property` ¶

recording_call_id: str

Call-ID of the SIPREC INVITE (the recording dialog).

original_call_id `property` ¶

original_call_id: str | None

Call-ID of the original call being recorded (from metadata).

participants `property` ¶

participants: list[SrsParticipant]

Participants in the recorded call.

duration `property` ¶

duration: int

Recording duration in seconds.

recording_dir `property` ¶

recording_dir: str | None

Directory where RTPEngine wrote the recording files.

`RecordingMetadata`¶

Parsed RFC 7866 recording metadata from a SIPREC INVITE.

Passed to @srs.on_invite handlers so the script can inspect participants, streams, and the session ID before accepting/rejecting.

Example::

@srs.on_invite
async def on_recording(metadata):
    if any(p.aor == "sip:vip@example.com" for p in metadata.participants):
        return True   # always record VIP calls
    return False      # reject others

session_id `property` ¶

session_id: str

Recording session ID from the SIPREC metadata.

participants `property` ¶

participants: list[SrsParticipant]

List of participants in the recorded call.

streams `property` ¶

streams: list[SrsStreamInfo]

List of media streams being recorded.

`SrsParticipant`¶

A participant in the recorded call.

Attributes:

Name	Type	Description
`participant_id`	`str`	Unique identifier for this participant.
`aor`	`str`	Address of Record (SIP URI, e.g. `"sip:alice@example.com"`).
`name`	`str \| None`	Optional display name.

participant_id `property` ¶

participant_id: str

Participant identifier from the recording metadata.

aor `property` ¶

aor: str

Address of Record (SIP URI).

name `property` ¶

name: str | None

Optional display name.

`SrsStreamInfo`¶

A media stream being recorded.

Attributes:

Name	Type	Description
`stream_id`	`str`	Unique identifier for this stream.
`label`	`str`	Stream label (correlates with SDP `a=label`).

stream_id `property` ¶

stream_id: str

Stream identifier from the recording metadata.

label `property` ¶

label: str

Stream label (e.g. "main-audio", "caller-audio").

IMS control¶

isc namespace¶

store_profile ¶

shared_set_ids ¶

remove_profile ¶

has_profile ¶

evaluate ¶

profile_count ¶

set_eval_results ¶

clear ¶

sbi namespace¶

PCF callbacks¶

set_notif_uri ¶

create_session ¶

delete_session async ¶

update_session ¶

discover_pcf_binding ¶

on_event staticmethod ¶

on_terminate staticmethod ¶

set_authorized ¶

set_binding ¶

set_bsf_error ¶

set_delete_failure ¶

clear ¶

BsfError¶

presence namespace¶

notifications property ¶

publish ¶

lookup ¶

subscribe ¶

subscribe_dialog ¶

unsubscribe ¶

refresh ¶

find_by_dialog ¶

next_version ¶

subscribers ¶

subscription_count ¶

document_count ¶

notify ¶

set_notify_response ¶

terminate ¶

parse_reginfo ¶

clear ¶

li namespace¶

is_enabled property ¶

task_count property ¶

destination_count property ¶

events property ¶

targets property ¶

is_target ¶

intercept ¶

record ¶

stop_intercept ¶

set_provisioned_counts ¶

stop_recording ¶

add_target ¶

clear ¶

srs namespace¶

accept_all property writable ¶

sessions property ¶

invite_events property ¶

on_invite ¶

on_session_end ¶

record_invite ¶

record_session_end ¶

clear ¶

SrsSession¶

session_id property ¶

recording_call_id property ¶

original_call_id property ¶

participants property ¶

duration property ¶

recording_dir property ¶

RecordingMetadata¶

session_id property ¶

participants property ¶

streams property ¶

SrsParticipant¶

participant_id property ¶

aor property ¶

`isc` namespace¶

`sbi` namespace¶

delete_session `async` ¶

on_event `staticmethod` ¶

on_terminate `staticmethod` ¶

`BsfError`¶

`presence` namespace¶

notifications `property` ¶

`li` namespace¶

is_enabled `property` ¶

task_count `property` ¶

destination_count `property` ¶

events `property` ¶

targets `property` ¶

`srs` namespace¶

accept_all `property` `writable` ¶

sessions `property` ¶

invite_events `property` ¶

`SrsSession`¶

session_id `property` ¶

recording_call_id `property` ¶

original_call_id `property` ¶

participants `property` ¶

duration `property` ¶

recording_dir `property` ¶

`RecordingMetadata`¶

session_id `property` ¶

participants `property` ¶

streams `property` ¶

`SrsParticipant`¶

participant_id `property` ¶

aor `property` ¶

name `property` ¶

`SrsStreamInfo`¶

stream_id `property` ¶

label `property` ¶