Skip to content

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.

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::

from siphon_sdk import mock_module
mock_module.install()

from siphon import sbi
result = 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,
) -> 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]

Notification URI for PCF events.

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

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.

delete_session

delete_session(session_id: str) -> bool

Delete an N5 app session.

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 on success, False if session not found.

update_session

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

Update an N5 app session (media renegotiation).

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

Returns:

Type Description
Optional[dict]

Dict with app_session_id and authorized, or None.

discover_pcf_binding

discover_pcf_binding(
    ue_ipv4: Optional[str] = None,
    ue_ipv6: Optional[str] = None,
) -> 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
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).

Example::

@sbi.on_event
def handle_pcf_event(event):
    session_events_uri = event.get("evSubsUri")
    for notif in event.get("evNotifs", []):
        log.info(f"PCF event: {notif['event']}")

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

clear

clear() -> None

Reset all mock sessions (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,
) -> str

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

Parameters:

Name Type Description Default
subscriber str

Watcher URI (Contact from the SUBSCRIBE).

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

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

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):
    li.intercept(request)

Test assertions::

li = get_li()
assert len(li.events) == 1
assert li.events[0] == ("intercept", "sip:alice@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.

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

Trigger interception for a matching request (emit IRI-BEGIN + start media capture).

Parameters:

Name Type Description Default
request Any

The SIP request object.

required

Returns:

Type Description
bool

True if interception was triggered for at least one matching target.

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.

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

Stop interception for a request (emit IRI-END).

Parameters:

Name Type Description Default
request Any

The SIP request object.

required

Returns:

Type Description
bool

True if a stop event was emitted for at least one matching target.

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.

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