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).

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.

In the mock, the XML is stored as-is (no actual parsing). 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`	Number of iFCs "parsed" (always 1 in mock unless configured otherwise).

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",
) -> list[dict]

Evaluate iFCs for a request.

Returns pre-configured results (via set_eval_results) or an empty list.

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`	Session case string.	`'originating'`

Returns:

Type	Description
`list[dict]`	List of dicts with keys: `server_name`, `default_handling`,
`list[dict]`	`service_info`, `priority`.

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`.	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.

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

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 for correlation.	`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`. 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`	`events` is empty, or given without `notif_uri`.
`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 (404 / 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/prefix.	`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` (404 / 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.

Bases: RuntimeError

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

A 404 (no binding for the UE IP) is not a BsfError — it 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()

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,
) -> 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.	`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`

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.

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",
) -> None

Send an in-dialog NOTIFY for a subscription.

In the mock, this records the notification for test assertions.

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.

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'`

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 ¶

remove_profile ¶

has_profile ¶

evaluate ¶

profile_count ¶

set_eval_results ¶

clear ¶

sbi namespace¶

PCF callbacks¶

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 ¶

subscribers ¶

subscription_count ¶

document_count ¶

notify ¶

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 ¶

name property ¶

SrsStreamInfo¶

stream_id property ¶

label 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` ¶