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_sdk — from 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:

API reference

Namespaces

Request properties

Request methods

Registrar

lookup() returns Contact objects:

Auth

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.

Provisioning outbound registrations

siphon_sdk.registrants is the typed mirror of the contract a registrant.backend: http source answers with — the list of trunks siphon should keep registered. A controller that owns trunks as data serves it and siphon reconciles against it, so a trunk added, edited or deleted needs no restart and no script:

from fastapi import FastAPI
from siphon_sdk.registrants import RegistrantListResponse, RegistrantRow

app = FastAPI()

@app.get("/registrants")
def registrants() -> dict:
    return RegistrantListResponse(registrants=[
        RegistrantRow(
            aor="sip:trunk1@carrier.example",
            registrar="sip:carrier.example:5060",
            username="trunk1",
            password="…",
        ),
    ]).to_dict()

The list is the complete desired state, not a delta: a trunk that is absent is de-registered. Supply password or ha1, never both — an ha1 is not reversible to the password but is still password-equivalent for its realm, and is bound to the hash it was computed with.

Full contract, including the SQL column names the database source reads: https://siphon-sip.org/reference/registrant-api/.

Provisioning gateways

siphon_sdk.gateways is the same idea for where calls egress to. One row is one destination; rows are gathered into groups by group, which is the name gateway.select() takes:

from siphon_sdk.gateways import GatewayListResponse, GatewayRow

@app.get("/gateways")
def gateways() -> dict:
    return GatewayListResponse(gateways=[
        GatewayRow(group="carriers", uri="sip:gw1.carrier.example:5060",
                   weight=3, registers="sip:trunk1@carrier.example"),
    ]).to_dict()

registers links a destination to an outbound registration: it answers challenges with that registration's credential, so a trunk's secret is defined once, and require_registration=True additionally keeps it out of selection while that registration is down.

A destination whose definition has not changed keeps its health across a refresh, so a carrier that is down stays down rather than being marked healthy again every poll.

Full contract: https://siphon-sip.org/reference/gateway-api/.

Consuming CDRs

siphon_sdk.cdr.CallDetailRecord is the typed mirror of the record siphon writes to its CDR sinks — one JSON object per HTTP POST, per line of the JSON-lines file, per syslog message. Import it in your collector instead of hand-parsing dicts:

from fastapi import FastAPI
from siphon_sdk.cdr import CallDetailRecord

app = FastAPI()

@app.post("/cdr")
async def collect(payload: dict) -> dict:
    record = CallDetailRecord.from_dict(payload)
    if record.is_media:                       # method == "MEDIA"
        for leg in record.media_legs:         # join to the call on call_id
            print(leg.role, leg.codec, leg.packets_lost)
    else:
        print(record.call_id, record.duration_secs, record.reason_cause)
        print(record.extra["billing_id"])     # cdr.write(extra={...})
    return {"ok": True}

Take the body as dict and call from_dict. Custom fields are flattened into the top level of the JSON, and a validating body model drops them before the handler runs; from_dict keeps them in record.extra.

examples/cdr_collector.py in the SIPhon repo is a runnable version of this.

License

MIT