Skip to main content

SIPhon

siphon-sip

Mock library and type stubs for SIPhon scripts — enables unit testing without the Rust binary and provides rich context for LLM-assisted script authoring.

Install

pip install siphon-sip

The PyPI distribution is siphon-sip (matching the siphon-sip crate). The import package is still siphon_sdkfrom siphon_sdk import ….

What is SIPhon?

SIPhon is a high-performance SIP proxy, B2BUA, and IMS platform written in Rust with Python scripting. Scripts use decorators to handle SIP events:

from siphon import proxy, registrar, auth, log

@proxy.on_request
def route(request):
    if request.method == "REGISTER":
        if not auth.require_digest(request, realm="example.com"):
            return
        registrar.save(request)
        request.reply(200, "OK")
        return

    contacts = registrar.lookup(request.ruri)
    if not contacts:
        request.reply(404, "Not Found")
        return

    request.record_route()
    request.fork([c.uri for c in contacts])

This SDK lets you test these scripts with pytest — no Rust binary needed.

Quick start

from siphon_sdk.testing import SipTestHarness
from siphon_sdk.types import Contact

harness = SipTestHarness(local_domains=["example.com"])
harness.load_script("scripts/proxy_default.py")

# Pre-populate the registrar
harness.registrar.add_contact(
    "sip:alice@example.com",
    Contact(uri="sip:alice@192.168.1.5:5060"),
)

# Test REGISTER challenge
result = harness.send_request("REGISTER", "sip:alice@example.com",
                              from_uri="sip:alice@example.com")
assert result.status_code == 401  # digest challenge

# Test INVITE routing
result = harness.send_request("INVITE", "sip:alice@example.com")
assert result.action == "fork"
assert "sip:alice@192.168.1.5:5060" in result.targets

Testing B2BUA scripts

harness = SipTestHarness()
harness.load_script("scripts/b2bua_default.py")

harness.registrar.add_contact(
    "sip:bob@example.com",
    Contact(uri="sip:bob@10.0.0.2:5060"),
)

result = harness.send_invite(ruri="sip:bob@example.com")
assert result.action == "fork"
assert result.targets == ["sip:bob@10.0.0.2:5060"]

# Test BYE handling
result = harness.send_bye(initiator_side="a")
assert result.was_terminated

Testing extension scripts (SMPP, HTTP)

The opt-in siphon extensions inject extra namespaces (smpp, http) at runtime. The SDK mocks them too, with dedicated harnesses, so extension scripts are testable from the same pip install siphon-sip — no running SMSC or HTTP listener required:

from siphon_sdk.smpp_testing import SmppTestHarness

harness = SmppTestHarness()
harness.load_script("scripts/gateway.py")
assert harness.bind("esme1", password="s3cret")
reply = harness.submit_sm(source_addr="15550100", destination_addr="15550101",
                          short_message=b"hi")
assert reply.ok
from siphon_sdk.http_testing import HttpTestHarness
from siphon_sdk.http import MockResponse

harness = HttpTestHarness()
harness.add_response(MockResponse(status=200, body=b'{"ok":true}'))
harness.load_script("scripts/api.py")

resp = harness.request("GET", "/users/42")
assert resp.status == 200

Inline scripts

Test scripts without separate files:

harness = SipTestHarness()
harness.load_source("""
from siphon import proxy

@proxy.on_request
def route(request):
    if request.source_ip_in(["10.0.0.0/8"]):
        request.relay()
    else:
        request.reply(403, "Forbidden")
""")

result = harness.send_request("INVITE", "sip:bob@host", source_ip="10.1.2.3")
assert result.was_relayed

result = harness.send_request("INVITE", "sip:bob@host", source_ip="8.8.8.8")
assert result.status_code == 403

Async handlers + RTPEngine

harness = SipTestHarness()
harness.load_source("""
from siphon import proxy, rtpengine

@proxy.on_request
async def route(request):
    if request.method == "INVITE" and request.body:
        await rtpengine.offer(request, profile="srtp_to_rtp")
    request.relay()
""")

result = harness.send_request("INVITE", "sip:bob@host",
                              body=b"v=0\\r\\n...",
                              content_type="application/sdp")
assert result.was_relayed
assert harness.rtpengine.operations == [("offer", "srtp_to_rtp")]

Controlling mock behavior

# Auth: allow or deny all
harness.auth._allow = True  # all auth checks pass

# Rate limiting
harness.proxy._utils._rate_limit_allow = False  # simulate overload

# Cache: pre-populate
harness.cache.set_data("cnam", {"key": "value"})

# Registrar: add contacts directly
harness.registrar.add_contact("sip:alice@host", Contact(uri="sip:alice@1.2.3.4"))

# Log: inspect captured messages
assert any("error" in msg for level, msg in harness.log.messages)

# Reset between tests
harness.reset()

Result assertions

RequestResult provides convenient properties:

Property Description
.action Primary action: "reply", "relay", "fork", "silent_drop"
.status_code SIP status code (200, 401, 404, etc.)
.reason Reason phrase
.targets Fork targets list
.strategy Fork strategy ("parallel" / "sequential")
.was_relayed True if relay() was called
.was_forked True if fork() was called
.was_dropped True if handler returned without action (silent drop)
.record_routed True if record_route() was called
.request The mock Request object for header inspection

API reference

Namespaces

Import Description
proxy Stateful/stateless proxy decorators and utilities
registrar Address-of-record contact store
auth SIP digest authentication
b2bua Back-to-back user agent call control
log Structured logging
cache Named cache (local LRU + Redis)
rtpengine RTPEngine media proxy operations
gateway Destination groups, load balancing, health probing
cdr Call detail records
diameter Diameter protocol (Cx, Ro, Rx, Rf, Sh)
presence SUBSCRIBE/NOTIFY, PIDF presence
li Lawful intercept (ETSI X1/X2/X3, SIPREC)
registration Outbound REGISTER client (trunk registration)

Request properties

Property Type Description
method str SIP method ("INVITE", "REGISTER", etc.)
ruri SipUri Request-URI
from_uri SipUri | None From header URI
to_uri SipUri | None To header URI
from_tag str | None From-tag
to_tag str | None To-tag (None for initial requests)
call_id str | None Call-ID
cseq (int, str) | None CSeq tuple
in_dialog bool Both tags present
max_forwards int Max-Forwards value
body bytes | None Message body
content_type str | None Content-Type
transport str "udp", "tcp", "tls", "ws", "wss"
source_ip str Sender IP
auth_user str | None Authenticated username
event str | None Event header

Request methods

Method Description
reply(code, reason) Send SIP response
relay(next_hop=None) Forward to destination
fork(targets, strategy="parallel") Fork to multiple targets
record_route() Insert Record-Route
loose_route() -> bool RFC 3261 loose routing
get_header(name) -> str | None Get header value
set_header(name, value) Set header
remove_header(name) Remove header
has_header(name) -> bool Check header exists
has_body(content_type) -> bool Check body type
set_ruri_user(value) Set R-URI user part
set_ruri_host(value) Set R-URI host
source_ip_in(cidrs) -> bool CIDR membership check
generate_icid() -> str Generate charging ID
add_path(uri) Prepend Path header
prepend_route(uri) Prepend Route header
fix_nated_register() NAT fixup for REGISTER
fix_nated_contact() NAT fixup for Contact

Registrar

Method Description
save(request, force=False) Save REGISTER bindings
lookup(uri) -> list[Contact] Look up contacts (sorted by q-value)
is_registered(uri) -> bool Check if URI has contacts
service_route(uri) -> list[str] Get stored service routes (RFC 3608)
set_service_routes(aor, routes) Store service routes for an AoR
save_pending(request) IMS: save binding in pending state
confirm_pending(uri) IMS: promote pending to active after SAR
asserted_identity(uri) -> str | None IMS: stored P-Asserted-Identity
reginfo_xml(aor, state, version) -> str Generate reginfo XML (RFC 3680)
on_change Decorator: fires on registration state changes

Auth

Method Description
require_www_digest(request, realm) -> bool 401 challenge
require_proxy_digest(request, realm) -> bool 407 challenge
require_digest(request, realm) -> bool Alias for www_digest
verify_digest(request, realm) -> bool Verify without challenge
require_ims_digest(request, realm) -> bool IMS AKA via Diameter Cx MAR
require_aka_digest(request, realm) -> bool Local Milenage AKA (no HSS)

B2BUA call

Each B-leg gets a fresh Call-ID and From-tag by default, fully decoupling the two SIP dialogs. Use keep_call_id() to opt out of Call-ID regeneration.

Property/Method Description
call.id UUID
call.state "calling", "ringing", "answered", "terminated"
call.from_uri A-leg From URI
call.ruri A-leg Request-URI
call.reject(code, reason) Reject call
call.dial(uri, timeout=30) Dial single target
call.fork(targets, strategy, timeout) Fork to multiple
call.terminate() End call (BYE both legs)
call.keep_call_id() Copy A-leg Call-ID to B-leg (From-tag always unique)
call.set_credentials(user, pass) B-leg digest auth credentials (auto 401/407 retry)
call.media.anchor(engine) Anchor media through RTPEngine
call.media.release() Release media anchor
call.session_timer(expires, min_se, refresher) Per-call RFC 4028 session timer
call.record(srs_uri) Start SIPREC recording
call.stop_recording() Stop SIPREC recording

License

MIT

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

siphon_sip-1.3.0.tar.gz (144.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

siphon_sip-1.3.0-py3-none-any.whl (118.9 kB view details)

Uploaded Python 3

File details

Details for the file siphon_sip-1.3.0.tar.gz.

File metadata

  • Download URL: siphon_sip-1.3.0.tar.gz
  • Upload date:
  • Size: 144.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for siphon_sip-1.3.0.tar.gz
Algorithm Hash digest
SHA256 78a980efc1afdbf6b882dfa96ff872a801966df2c009ba1b5b931636efd488ae
MD5 49b375ac8355f7732611c7edd08f415d
BLAKE2b-256 680b6493582188e096c2a5b40413b2aa2268a53d8ce147e012cd1d0f5f9f16f7

See more details on using hashes here.

Provenance

The following attestation bundles were made for siphon_sip-1.3.0.tar.gz:

Publisher: release.yaml on siphon-project/siphon-sip

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file siphon_sip-1.3.0-py3-none-any.whl.

File metadata

  • Download URL: siphon_sip-1.3.0-py3-none-any.whl
  • Upload date:
  • Size: 118.9 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for siphon_sip-1.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 d02bc138afdb2672321f0b004b4768e6ed86ddb1b3823930ebce4f83672fbef2
MD5 ab42ad9ddc97fd2e00307ad57ed60f1d
BLAKE2b-256 98663bef57252a690f954bcaf2627fd678bff904b8ed04ecf8965cb0a60c5d72

See more details on using hashes here.

Provenance

The following attestation bundles were made for siphon_sip-1.3.0-py3-none-any.whl:

Publisher: release.yaml on siphon-project/siphon-sip

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Release history Release notifications | RSS feed

1.6.0

2 files

1.5.1

2 files

1.5.0

2 files

1.4.1

2 files

1.4.0

2 files

This release

1.3.0 This release

2 files

1.2.1

2 files

1.2.0

2 files

1.1.1

2 files

1.1.0

2 files

1.0.0

2 files

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page