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
¶
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 a stored profile.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aor
|
str
|
Address of Record. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
has_profile
¶
Check if a profile is stored for an AoR.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
aor
|
str
|
Address of Record. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
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. |
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: |
list[dict]
|
|
set_eval_results
¶
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 |
required |
Example::
isc.set_eval_results("sip:alice@example.com", [
{"server_name": "sip:as1@example.com", "default_handling": 0,
"service_info": None, "priority": 0},
])
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'
|
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
|
None
|
pcf_uri
|
Optional[str]
|
per-call N5 target — address this session at the given PCF
base URL (e.g. a BSF-discovered |
None
|
Returns:
| Type | Description |
|---|---|
Optional[dict]
|
Dict with |
Optional[dict]
|
(the absolute resource URI — persist it and hand it back to |
Optional[dict]
|
|
Optional[dict]
|
teardown), or |
delete_session
¶
Delete an N5 app session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session_id
|
str
|
The app session id from |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
update_session
¶
Update an N5 app session (media renegotiation).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session_id
|
str
|
The app session id to update, or the absolute
|
required |
media_components
|
Optional[list]
|
list of media-component dicts (same shape as
|
None
|
Returns:
| Type | Description |
|---|---|
Optional[dict]
|
Dict with |
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 |
on_event
staticmethod
¶
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
¶
Configure whether create_session returns authorized (test helper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
authorized
|
bool
|
Whether sessions should be authorized. |
required |
set_binding
¶
Configure what discover_pcf_binding returns (test helper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
binding
|
Optional[dict]
|
a binding dict (5G case) or |
required |
set_bsf_error
¶
Configure discover_pcf_binding to raise BsfError (test helper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
raise_error
|
bool
|
when True, |
required |
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()
publish
¶
Publish a presence document for a presentity.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
entity
|
str
|
Presentity URI (e.g. |
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
¶
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 |
subscribe
¶
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. |
required |
resource
|
str
|
Presentity URI to watch. |
required |
event
|
str
|
Event package name (default: |
'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 by subscription ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
subscription_id
|
str
|
The subscription ID returned by :meth: |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
refresh
¶
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 |
required |
expires
|
int
|
New subscription duration in seconds. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
find_by_dialog
¶
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 |
subscribers
¶
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: |
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 |
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'
|
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 |
required |
reason
|
Optional[str]
|
Termination reason per RFC 6665 §4.2.2 — one of
|
None
|
body
|
Optional[str]
|
Optional final body. |
None
|
content_type
|
Optional[str]
|
Content-Type of the body. |
None
|
Returns:
| Type | Description |
|---|---|
bool
|
|
bool
|
recorded; |
Example::
sub_id = presence.subscribe_dialog(...)
...
presence.terminate(sub_id, reason="timeout")
parse_reginfo
¶
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.
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
¶
Whether the LI subsystem is enabled.
In the mock, returns True if _enabled is set and targets
are configured.
events
property
¶
List of (operation, target_or_call_id) tuples recorded.
Operations: "intercept", "record", "stop_intercept",
"stop_recording".
is_target
¶
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
|
|
intercept
¶
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
|
|
record
¶
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 |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
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 interception for a request (emit IRI-END).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
request
|
Any
|
The SIP request object. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
stop_recording
¶
Stop SIPREC recording for a request or call.
Accepts either a Request or Call object.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target
|
Any
|
A |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
add_target
¶
Add a target URI for intercept matching (test helper).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
uri
|
str
|
SIP URI to match against (e.g. |
required |
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
¶
Whether mock auto-accepts all recordings (default True).
sessions
property
¶
List of completed recording sessions (for test assertions).
invite_events
property
¶
List of on_invite calls received (for test assertions).
on_invite
¶
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
¶
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
¶
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
|
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}")
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
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. |
name |
str | None
|
Optional display name. |