class RfcServer:
"""Sans-I/O RFC server: handler registry + inbound dispatch + asyncio serve.
Usage (sync facade, D-02)::
server = RfcServer({"program_id": "MY_PROG", "gwhost": "gw", "gwserv": "sapgw00"})
@server.function("STFC_CONNECTION", stfc_desc)
def handle(request: dict) -> dict:
return {"ECHOTEXT": request["REQUTEXT"], "RESPTEXT": "ok"}
server.serve_forever() # blocks; runs the asyncio loop in a daemon thread
# ... server.stop() from another thread to tear down cleanly ...
The offline-testable core is ``dispatch_inbound(frame) -> bytes`` (frame bytes
in, response bytes out) — no sockets. ``serve``/``serve_forever`` drive it over
a live gateway connection.
"""
def __init__(self, params: dict[str, Any]) -> None:
# Registration params (program_id/gwhost/gwserv). Shape is discretion; the
# ServerSession enforces the registration constraints when serve() registers.
self._params = dict(params)
# FM_NAME.upper() -> (FunctionDesc | None, handler) — mirrors MetadataCache keying.
# func_desc may be None for transactional handlers where dispatch short-circuits
# before deserialization (no FunctionDesc needed at registration time).
self._registry: dict[
str, tuple[FunctionDesc | None, Callable[[dict[str, Any]], dict[str, Any]]]
] = {}
# Generic fallback consulted on a registry miss (D-09); None by default.
self._generic: (
Callable[
[str], tuple[FunctionDesc | None, Callable[[dict[str, Any]], dict[str, Any]]] | None
]
| None
) = None
# Auth callback (SERVER-05); None means "no auth check" (allow all).
self._auth_check: Callable[..., bool] | None = None
# TID store (TRFC-03 / D-03): default InMemoryTidStore so a bare server works.
# Replace with a custom durable store via set_tid_store().
self._tid_store: TidStore = InMemoryTidStore()
# Transaction lifecycle callbacks (SDK type definitions-732 — RfcInstallTransactionHandlers).
# When set, invoked at the corresponding check→commit/rollback→confirm sequence points.
# Unset (None) means the store-only default behaviour applies (Task 1 dispatch).
self._on_check_transaction: Callable[[str], int] | None = None
self._on_commit_transaction: Callable[[str], None] | None = None
self._on_rollback_transaction: Callable[[str], None] | None = None
self._on_confirm_transaction: Callable[[str], None] | None = None
# bgRFC Unit store (TRFC-07 / D-03): default InMemoryUnitStore so a bare server works.
# Replace with a custom durable store via set_unit_store().
self._unit_store: UnitStore = InMemoryUnitStore()
# bgRFC unit lifecycle callbacks (SDK type definitions-741 — RfcInstallBgRfcHandlers).
# Unset (None) means store-only default behaviour applies.
self._on_check_unit: Callable[..., int] | None = None
self._on_commit_unit: Callable[..., None] | None = None
self._on_rollback_unit: Callable[..., None] | None = None
self._on_confirm_unit: Callable[..., None] | None = None
self._on_get_unit_state: Callable[..., UnitState] | None = None
# serve-loop lifecycle (set up by serve_forever).
self._thread: threading.Thread | None = None
self._transport: Transport | None = None # live Transport; set during _serve_blocking
self._stopped: bool = False
self._session = ServerSession()
# --------------------------------------------------------------------- #
# Registry + decorator (SERVER-01, D-07/D-08/D-09)
# --------------------------------------------------------------------- #
def function(
self, name: str, func_desc: FunctionDesc | None = None
) -> Callable[
[Callable[[dict[str, Any]], dict[str, Any]]], Callable[[dict[str, Any]], dict[str, Any]]
]:
"""Decorator registering ``(func_desc, fn)`` under ``name.upper()`` (D-07).
``func_desc`` is optional: for transactional handlers where the server
dispatch short-circuits before deserialization (e.g. duplicate TID returns
RFC_EXECUTED without deserializing the request), ``FunctionDesc`` is not
needed at registration time. Pass ``None`` or omit it in that case.
The wrapped handler is returned unchanged, so the decorated name stays a
normal callable::
@server.function("STFC_CONNECTION", desc) # sync: desc required for deserialization
def handle(request: dict) -> dict: ...
@server.function("MY_TRFC_FM") # transactional: no desc needed at register
def trfc_handle(request: dict) -> dict: ...
"""
key = name.upper()
def _register(
fn: Callable[[dict[str, Any]], dict[str, Any]],
) -> Callable[[dict[str, Any]], dict[str, Any]]:
self._registry[key] = (func_desc, fn)
return fn
return _register
def set_generic_handler(
self,
fn: Callable[[str], tuple[FunctionDesc, Callable[[dict[str, Any]], dict[str, Any]]] | None],
) -> None:
"""Register a fallback consulted on a registry miss (D-09).
``fn(func_name)`` returns ``(FunctionDesc, handler)`` to serve the call, or
``None`` to decline (→ SYSTEM_FAILURE). The default is no generic handler,
so an unknown FM yields SYSTEM_FAILURE with no information leak about which
FMs are registered (threat V4 / T-05-C05 — generic handler is explicit
opt-in and must validate/whitelist).
"""
self._generic = fn
def set_authentication_check(self, fn: Callable[..., bool]) -> None:
"""Register an auth callback run before handler dispatch (SERVER-05).
``fn`` receives the inbound credentials and returns ``True`` to allow the
call or ``False`` to deny it. On deny the handler is NOT invoked and an
auth-failure SYSTEM_FAILURE is returned (T-05-C06). The callback signature
is invoked as ``fn(user=..., password=...)`` (keyword args); a single
unhandled credential field is passed as ``None``. Inbound credentials are
NEVER logged (T-04-CRED / T-05-C03).
"""
self._auth_check = fn
def set_tid_store(self, store: TidStore) -> None:
"""Replace the default InMemoryTidStore with a custom durable store (TRFC-08).
The custom store is used by ``dispatch_inbound`` for the transactional
dispatch branch: duplicate-TID detection (``is_executed``), crash-safe
persistence (``mark_received`` before handler), and lifecycle management
(``mark_executed`` / ``mark_rolled_back`` / ``confirm``).
``store`` must satisfy the :class:`~saprfclib.stores.TidStore` Protocol
(structural typing, D-01). Optionally validated with isinstance when
TidStore is @runtime_checkable.
Example::
db_store = MyPostgreSQLTidStore(conn)
server.set_tid_store(db_store)
"""
if not isinstance(store, TidStore):
raise TypeError(
f"store must implement the TidStore Protocol, got {type(store).__name__!r}"
)
self._tid_store = store
def install_transaction_handlers(
self,
*,
on_check: Callable[[str], int] | None = None,
on_commit: Callable[[str], None] | None = None,
on_rollback: Callable[[str], None] | None = None,
on_confirm: Callable[[str], None] | None = None,
) -> None:
"""Register the four tRFC transaction lifecycle callbacks (TRFC-03).
Maps the ``RfcInstallTransactionHandlers`` API (SDK type definitions-732):
- ``on_check(tid) -> int``: called first; return 0 (RFC_OK) for a new TID
or 16 (RFC_EXECUTED) if already executed. When set, this callback REPLACES
the default store-based check (``TidStore.is_executed``).
- ``on_commit(tid) -> None``: called after the handler succeeds; maps to
``TidStore.mark_executed``.
- ``on_rollback(tid) -> None``: called when the handler raises an exception;
maps to ``TidStore.mark_rolled_back``.
- ``on_confirm(tid) -> None``: called as cleanup after commit/rollback;
maps to ``TidStore.confirm``.
When unset (None), the corresponding store method is used directly. Setting
the callbacks enables the full ``RFC_ON_CHECK/COMMIT/ROLLBACK/CONFIRM_TRANSACTION``
contract from SDK type definitions.
Example::
server.install_transaction_handlers(
on_check=lambda tid: 16 if my_db.has(tid) else 0,
on_commit=lambda tid: my_db.commit(tid),
on_rollback=lambda tid: my_db.rollback(tid),
on_confirm=lambda tid: my_db.remove(tid),
)
"""
self._on_check_transaction = on_check
self._on_commit_transaction = on_commit
self._on_rollback_transaction = on_rollback
self._on_confirm_transaction = on_confirm
def set_unit_store(self, store: UnitStore) -> None:
"""Replace the default InMemoryUnitStore with a custom durable store (TRFC-08).
The custom store is used by ``dispatch_inbound`` for the bgRFC unit dispatch
branch: unit state tracking (``get_unit_state``), persistence (``persist``),
and lifecycle management (``confirm``).
``store`` must satisfy the :class:`~saprfclib.stores.UnitStore` Protocol
(structural typing, D-02). Optionally validated with isinstance when
UnitStore is @runtime_checkable.
Example::
db_store = MyPostgreSQLUnitStore(conn)
server.set_unit_store(db_store)
"""
if not isinstance(store, UnitStore):
raise TypeError(
f"store must implement the UnitStore Protocol, got {type(store).__name__!r}"
)
self._unit_store = store
def install_unit_handlers(
self,
*,
check: Callable[..., int] | None = None,
commit: Callable[..., None] | None = None,
rollback: Callable[..., None] | None = None,
confirm: Callable[..., None] | None = None,
get_state: Callable[..., UnitState] | None = None,
) -> None:
"""Register the five bgRFC unit lifecycle callbacks (TRFC-07).
Maps ``RfcInstallBgRfcHandlers`` (SDK type definitions-741):
- ``check(unit_id, unit_type) -> int``: called first; return 0 (RFC_OK) for a
new unit or 16 (RFC_EXECUTED) if already known.
- ``commit(unit_id, unit_type) -> None``: called after handler executes
successfully; maps to ``UnitStore.persist`` → ``UnitStore.confirm``.
- ``rollback(unit_id, unit_type) -> None``: called when a handler raises;
maps to ``UnitState.ROLLED_BACK`` state.
- ``confirm(unit_id, unit_type) -> None``: called after commit as cleanup;
maps to ``UnitStore.confirm``.
- ``get_state(unit_id, unit_type) -> UnitState``: called on inbound state
query; maps to ``UnitStore.get_unit_state``.
When unset (None), the corresponding store method is used directly.
Example::
server.install_unit_handlers(
check=lambda uid, ut: 16 if db.has_unit(uid) else 0,
commit=lambda uid, ut: db.commit_unit(uid, ut),
rollback=lambda uid, ut: db.rollback_unit(uid, ut),
confirm=lambda uid, ut: db.confirm_unit(uid, ut),
get_state=lambda uid, ut: db.get_unit_state(uid, ut),
)
"""
self._on_check_unit = check
self._on_commit_unit = commit
self._on_rollback_unit = rollback
self._on_confirm_unit = confirm
self._on_get_unit_state = get_state
# --------------------------------------------------------------------- #
# Sans-I/O dispatch core (SERVER-03/04/05/06) — frame bytes in, bytes out
# --------------------------------------------------------------------- #
def dispatch_inbound(self, frame: bytes) -> bytes:
"""Deserialize one inbound call, run the handler, serialize the response.
Pure function of ``frame`` (no I/O) — the offline-testable seam. Steps:
1. Strip the live GW header if present (``_strip_gw_header``; bare TLV from
MockTransport passes through — first byte != 0x06).
2. Read the function name (tag 0x0102, UTF-16LE).
Phase-6 Pitfall 6 seam: the call-type IS the function name.
Branch on name BEFORE normal handler lookup (TRFC-03):
- ``ARFC_DEST_SHIP`` → transactional tRFC/qRFC dispatch (this phase)
- ``ARFC_DEST_CONFIRM`` → tRFC confirm (Plan 06-05; placeholder SYSTEM_FAILURE)
- any other name → synchronous dispatch (existing path, unchanged)
Synchronous path (unchanged — Pitfall 2 regression guard):
3. Look up the handler, fall back to generic, then SYSTEM_FAILURE if unknown.
4. Auth check (SERVER-05) before handler.
5. Deserialize request (SERVER-03, Pitfall 4).
6. Handler dispatch, exception-isolated (SERVER-06).
7. Serialize response (SERVER-04).
"""
tlv = _strip_gw_header(frame)
# Phase-6 Pitfall 6 seam: function name IS the call-type discriminator
# (protocol analysis — no separate byte needed).
fname = self._read_func_name(tlv)
if fname == _ARFC_DEST_SHIP:
# Transactional tRFC/qRFC inbound dispatch (TRFC-03).
return self._dispatch_transactional(tlv, fname)
# ARFC_DEST_CONFIRM is handled by the gateway-side SAP infra; if it somehow
# reaches our registered server, respond with SYSTEM_FAILURE until a full
# confirm path is implemented (placeholder; does not regress sync tests).
if fname == _ARFC_DEST_CONFIRM:
return self._build_rfc_ok_response()
# bgRFC unit submit (TRFC-07): BGRFC_DEST_SHIP → _dispatch_bgrfc_unit
if fname == _BGRFC_DEST_SHIP:
return self._dispatch_bgrfc_unit(tlv)
# bgRFC confirm: BGRFC_DEST_CONFIRM → confirm the unit in the store
if fname == _BGRFC_DEST_CONFIRM:
return self._dispatch_bgrfc_confirm(tlv)
# bgRFC state query: BGRFC_CHECK_UNIT_STATE_SERVER → return UnitState
if fname == _BGRFC_CHECK_UNIT_STATE_SERVER:
return self._dispatch_bgrfc_state_query(tlv)
# ----- SYNCHRONOUS DISPATCH (unchanged — Pitfall 2 regression guard) -----
entry = self._registry.get(fname.upper())
if entry is None and self._generic is not None:
entry = self._generic(fname)
if entry is None:
# No info leak about the registered set (V4 / T-05-C05).
return self._build_system_failure(f"function {fname} not registered")
func_desc, handler = entry
# --- authentication (SERVER-05 / T-05-C06) — BEFORE handler dispatch ---
if self._auth_check is not None and not self._check_auth(tlv):
return self._build_system_failure("authentication failed")
# --- deserialize request → typed dict (SERVER-03, Pitfall 4) ---
request = self._deserialize_request(tlv, func_desc)
# --- handler dispatch, exception-isolated (SERVER-06 / D-03 / T-05-C01) ---
try:
result = handler(request)
except Exception as exc: # noqa: BLE001 — isolate ALL handler errors
# str(exc) only — never a full traceback, never credentials (T-05-C02).
return self._build_system_failure(str(exc))
if result is None:
result = {}
return self._build_response(func_desc, result)
def _dispatch_transactional(self, tlv: bytes, fname: str) -> bytes:
"""Transactional (tRFC/qRFC) inbound dispatch — TRFC-03 exactly-once gate.
Implements the server-side dedup state machine from RESEARCH Pattern 2
(docs/protocol/trfc.md §"Server-Side Dispatch") and SDK type definitions-732:
1. Extract TID from the ARFCTID param; validate (V5 — reject non-24-char
TIDs before any store call, T-06-D02).
2. check_transaction(tid): if is_executed → return RFC_EXECUTED (no handler).
3. mark_received(tid) BEFORE handler execute (crash-safety, T-06-D01).
4. Run the application handler inside the standard exception-isolation block
(reuse of lines 413-418 verbatim, T-06-D03 / T-05-C01).
a. Success → mark_executed(tid) + confirm(tid).
b. Exception → mark_rolled_back(tid) + SYSTEM_FAILURE(str(exc)) first-line only.
When ``install_transaction_handlers`` callbacks are set, they are invoked at
the corresponding points (step 2 → on_check; step 4a → on_commit + on_confirm;
step 4b → on_rollback). Store calls serve as fallback when callbacks are unset.
The synchronous dispatch path (other function names) is NOT touched.
"""
# --- TID extraction + validation (V5 / T-06-D02) ---
tid = self._extract_tid_from_frame(tlv)
if not tid:
return self._build_system_failure(
"ARFC_DEST_SHIP: missing ARFCTID param in inbound frame"
)
if len(tid) != _TID_LN:
return self._build_system_failure(
f"ARFC_DEST_SHIP: invalid TID length {len(tid)} (expected {_TID_LN})"
)
if any(c not in _TID_ALPHABET for c in tid):
return self._build_system_failure(
"ARFC_DEST_SHIP: TID contains characters outside RFC alphabet"
)
# --- check_transaction: is_executed? (TRFC-03 dedup short-circuit) ---
store = self._tid_store
if self._on_check_transaction is not None:
rc = self._on_check_transaction(tid)
already_done = rc == _RC_EXECUTED
else:
already_done = store.is_executed(tid)
if already_done:
# Known TID — return RFC_EXECUTED; DO NOT call the application handler.
return self._build_rfc_executed_response()
# --- persist BEFORE execute (crash-safety — Pattern 2, T-06-D01) ---
store.mark_received(tid)
# --- resolve handler from registry (wrapped FM name is in ARFCFNAM param) ---
wrapped_fn = self._extract_param_from_frame(tlv, "ARFCFNAM")
entry = None
if wrapped_fn:
entry = self._registry.get(wrapped_fn.upper())
if entry is None and self._generic is not None and wrapped_fn:
entry = self._generic(wrapped_fn)
# If no handler, run a no-op (consistent with exactly-once: we still mark TID).
# The handler must be present for a meaningful execution; if absent, the TID
# is persisted but SYSTEM_FAILURE is returned (deferred-handler path).
if entry is None:
store.mark_rolled_back(tid)
if self._on_rollback_transaction is not None:
self._on_rollback_transaction(tid)
return self._build_system_failure(
f"ARFC_DEST_SHIP: no handler registered for {wrapped_fn!r}"
)
func_desc, handler = entry
# --- deserialize request (Pitfall 4 direction flip — same as sync path) ---
request = self._deserialize_request(tlv, func_desc) if func_desc is not None else {}
# --- handler dispatch, exception-isolated (T-06-D03 / T-05-C01) ---
try:
result = handler(request)
except Exception as exc: # noqa: BLE001 — isolate ALL handler errors
store.mark_rolled_back(tid)
if self._on_rollback_transaction is not None:
self._on_rollback_transaction(tid)
# str(exc) only — no traceback, no credential leak (T-06-D03 / T-05-C02).
return self._build_system_failure(str(exc))
# --- commit (on_commit) — mark TID executed; keep in store for dedup ---
# NOTE: confirm() / on_confirm are NOT called here. The TID must remain
# in the store as "executed" so that retry deliveries see it as a dup
# and return RFC_EXECUTED (exactly-once guarantee). Confirmation (cleanup)
# happens only when the client sends ARFC_DEST_CONFIRM (a separate call),
# which maps to the on_confirm callback. Removing the TID here (calling
# store.confirm()) would break dedup on retry — Pitfall 3.
store.mark_executed(tid)
if self._on_commit_transaction is not None:
self._on_commit_transaction(tid)
if result is None:
result = {}
# tRFC has no EXPORTING params by design (CONTEXT Claude's discretion).
# Return a minimal success response (RFC_OK) without deserializing the desc.
return self._build_rfc_ok_response()
# --------------------------------------------------------------------- #
# bgRFC unit dispatch (TRFC-07) — Plan 06-05
# --------------------------------------------------------------------- #
def _dispatch_bgrfc_unit(self, tlv: bytes) -> bytes:
"""bgRFC unit inbound dispatch — TRFC-07 unit callback state machine.
Implements the server-side unit processing sequence from
docs/protocol/trfc.md §"Server-Side Dispatch" and SDK type definitions-2500:
1. Extract + validate UnitID (exactly 32 uppercase hex chars, V5 / T-06-U02).
2. Extract unit_type ('T' or 'Q') from frame params.
3. check_unit(uid, unit_type): if already executed → return RFC_EXECUTED.
4. persist(uid, unit_type) BEFORE handler execute (crash-safety, T-06-U01).
5. Execute each buffered call from the frame (exception-isolated, T-06-U03).
6. Success → on_commit(uid, unit_type) + store state = COMMITTED.
7. on_confirm(uid, unit_type) + store.confirm(uid, unit_type).
8. Exception → on_rollback(uid, unit_type) + store state = ROLLED_BACK.
Threat mitigations:
T-06-U01: persist before execute; confirm is a separate step.
T-06-U02: reject non-32-char or non-hex UnitID before store lookup.
T-06-U03: handler exception isolation — SYSTEM_FAILURE(str(exc)) only.
T-06-U04: NOT_FOUND after confirm is success; never resend (N/A server side).
"""
# --- UnitID extraction + validation (V5 / T-06-U02) ---
unit_id = self._extract_param_from_frame(tlv, "BGRFC_UNIT_ID")
if not unit_id:
return self._build_system_failure(
"BGRFC_DEST_SHIP: missing BGRFC_UNIT_ID param in inbound frame"
)
if len(unit_id) != _UNITID_LN:
return self._build_system_failure(
f"BGRFC_DEST_SHIP: invalid UnitID length {len(unit_id)} (expected {_UNITID_LN})"
)
if any(c not in _UNITID_CHARSET for c in unit_id):
return self._build_system_failure(
"BGRFC_DEST_SHIP: UnitID contains characters outside uppercase hex charset "
"(allowed: 0-9A-F)"
)
# --- unit_type extraction ---
unit_type = self._extract_param_from_frame(tlv, "BGRFC_UNIT_TYPE") or "T"
if unit_type not in ("T", "Q"):
unit_type = "T" # defensive default; invalid type treated as 'T'
unit_store = self._unit_store
# --- check_unit: already executed? ---
if self._on_check_unit is not None:
rc = self._on_check_unit(unit_id, unit_type)
already_done = rc == _RC_EXECUTED
else:
state = unit_store.get_unit_state(unit_id, unit_type)
already_done = state in (UnitState.COMMITTED, UnitState.CONFIRMED)
if already_done:
return self._build_rfc_executed_response()
# --- persist BEFORE execute (crash-safety, T-06-U01) ---
unit_store.persist(unit_id, unit_type)
# --- execute buffered calls (exception-isolated, T-06-U03) ---
# Each buffered call in the frame was embedded by build_bgrfc_request as raw
# bytes under BGRFC_CALL_N params. A unit is one LUW: the calls in it either
# all run or none of them count, and the caller re-ships the whole unit after
# a failure. So execution stops at the first error rather than carrying on --
# running the remaining calls would double-execute them on the resend.
call_error: str | None = None
call_count_str = self._extract_param_from_frame(tlv, "BGRFC_CALL_COUNT")
call_count = 0
if call_count_str:
try:
call_count = int(call_count_str)
except ValueError:
# Treating an unreadable count as zero commits the unit having run
# nothing, and reports that to the caller as a completed LUW.
return self._build_system_failure(
f"BGRFC_DEST_SHIP: unreadable BGRFC_CALL_COUNT {call_count_str!r}"
)
if call_count < 0:
return self._build_system_failure(
f"BGRFC_DEST_SHIP: negative BGRFC_CALL_COUNT {call_count}"
)
for i in range(call_count):
call_bytes = self._extract_raw_param_from_frame(tlv, f"BGRFC_CALL_{i}")
if call_bytes is None:
# The frame declared more calls than it carries. Skipping the gap
# would commit a partial LUW as a complete one.
call_error = (
f"bgRFC: frame declares {call_count} call(s) but BGRFC_CALL_{i} is missing"
)
break
# Try to decode the embedded call (func_name from UTF-16LE until NUL NUL).
try:
call_error = self._execute_buffered_call(call_bytes)
except Exception as exc: # noqa: BLE001 — isolate ALL errors (T-06-U03)
call_error = str(exc).splitlines()[0][:512]
if call_error is not None:
break
if call_error is not None:
# Exception in a unit call → rollback path (T-06-U03)
if self._on_rollback_unit is not None:
try:
self._on_rollback_unit(unit_id, unit_type)
except Exception: # noqa: BLE001 — a bad callback must not kill the server
# Isolated on purpose, but never silently: a rollback handler that
# throws has left the caller's own state half-undone, and that is
# the one thing nobody finds out about later.
_logger.exception(
"bgRFC: on_rollback_unit raised for unit %s (type %s); the unit "
"was rolled back on the wire but the callback did not complete",
unit_id,
unit_type,
)
return self._build_system_failure(call_error)
# --- success path → on_commit + store committed + on_confirm ---
if self._on_commit_unit is not None:
try:
self._on_commit_unit(unit_id, unit_type)
except Exception: # noqa: BLE001 — isolate callback errors
# Commit callback error: still confirm store (persist-then-commit
# separation). Logged because the unit is about to be confirmed as
# done while the caller's commit handler did not finish.
_logger.exception(
"bgRFC: on_commit_unit raised for unit %s (type %s); the unit is "
"being confirmed anyway (persist-then-commit separation)",
unit_id,
unit_type,
)
unit_store.confirm(unit_id, unit_type)
if self._on_confirm_unit is not None:
try:
self._on_confirm_unit(unit_id, unit_type)
except Exception: # noqa: BLE001 — cleanup must not fail the unit
_logger.exception(
"bgRFC: on_confirm_unit raised for unit %s (type %s)",
unit_id,
unit_type,
)
return self._build_rfc_ok_response()
def _dispatch_bgrfc_confirm(self, tlv: bytes) -> bytes:
"""Handle BGRFC_DEST_CONFIRM: confirm unit in the store."""
unit_id = self._extract_param_from_frame(tlv, "BGRFC_UNIT_ID")
unit_type = self._extract_param_from_frame(tlv, "BGRFC_UNIT_TYPE") or "T"
if not unit_id or len(unit_id) != _UNITID_LN:
return self._build_system_failure(
"BGRFC_DEST_CONFIRM: invalid or missing BGRFC_UNIT_ID"
)
# T-06-U04: NOT_FOUND after confirm = success; do not error.
self._unit_store.confirm(unit_id, unit_type)
if self._on_confirm_unit is not None:
try:
self._on_confirm_unit(unit_id, unit_type)
except Exception: # noqa: BLE001 — cleanup must not fail the unit
_logger.exception(
"bgRFC: on_confirm_unit raised for unit %s (type %s)",
unit_id,
unit_type,
)
return self._build_rfc_ok_response()
def _dispatch_bgrfc_state_query(self, tlv: bytes) -> bytes:
"""Handle BGRFC_CHECK_UNIT_STATE_SERVER: return unit state."""
unit_id = self._extract_param_from_frame(tlv, "BGRFC_UNIT_ID")
unit_type = self._extract_param_from_frame(tlv, "BGRFC_UNIT_TYPE") or "T"
if not unit_id or len(unit_id) != _UNITID_LN:
return self._build_system_failure(
"BGRFC_CHECK_UNIT_STATE_SERVER: invalid or missing BGRFC_UNIT_ID"
)
if self._on_get_unit_state is not None:
try:
state = self._on_get_unit_state(unit_id, unit_type)
except Exception as exc: # noqa: BLE001
# Answering NOT_FOUND here is the worst available answer: it tells the
# caller the unit was never seen, so the caller ships it again — and if
# it had in fact committed, the LUW runs twice. A failed lookup is not
# an absent unit; report that we do not know.
_logger.exception(
"bgRFC: on_get_unit_state raised for unit %s (type %s)",
unit_id,
unit_type,
)
return self._build_system_failure(
f"BGRFC_CHECK_UNIT_STATE_SERVER: state lookup for {unit_id} failed "
f"({type(exc).__name__}: {str(exc).splitlines()[0][:200]}) — the "
f"state is unknown, not NOT_FOUND"
)
else:
state = self._unit_store.get_unit_state(unit_id, unit_type)
# Encode state name as a CHAR param in the response TLV.
return b"".join(
[
tlv_record(_TAG_RESPONSE_START),
tlv_record(_TAG_RETURN_CODE, struct.pack(">I", _RC_OK)),
tlv_record(_TAG_PARAM_NAME, "BGRFC_STATE".encode("utf-16-le")),
tlv_record(_TAG_PARAM_VALUE, state.name.encode("utf-16-le")),
tlv_record(_TAG_TERMINATOR),
]
)
def _execute_buffered_call(self, call_bytes: bytes) -> str | None:
"""Execute one buffered call from a bgRFC unit payload.
Decodes the func_name from the call_bytes (UTF-16LE until NUL NUL separator),
looks up the handler, and dispatches it. Returns None on success or an error
string (str(exc) first line) on failure.
This method is exception-isolated: the caller wraps it in try/except to
satisfy T-06-U03 (no traceback leak, no credential leak).
"""
# Returning None here means "this call succeeded" to the caller, which then
# commits and confirms the unit. A call that could not be run is not a call
# that ran, so every unexecutable case below reports an error instead.
if not call_bytes:
return "bgRFC: buffered call is empty"
# Decode func_name: the leading UTF-16LE string up to its NUL terminator.
#
# The terminator is a 0x0000 *code unit*, so it can only start at an even
# offset. Scanning with bytes.find(b"\x00\x00") instead matched the low NUL
# of the last character plus the first NUL of the terminator -- an odd offset,
# every time, for any ASCII name. The odd result was then rejected as
# unaligned and the whole payload taken as the name, so the separator branch
# never ran and a call carrying parameters decoded to a garbage name.
nul_pos = -1
for off in range(0, len(call_bytes) - 1, 2):
if call_bytes[off] == 0 and call_bytes[off + 1] == 0:
nul_pos = off
break
try:
if nul_pos < 0:
# No terminator — the whole payload is the name.
func_name = call_bytes.decode("utf-16-le").rstrip("\x00 ")
else:
func_name = call_bytes[:nul_pos].decode("utf-16-le").rstrip("\x00 ")
except UnicodeDecodeError as exc:
return f"bgRFC: cannot decode function name from buffered call ({exc})"
if not func_name:
return "bgRFC: buffered call carries an empty function name"
entry = self._registry.get(func_name.upper())
if entry is None and self._generic is not None:
entry = self._generic(func_name)
if entry is None:
# No handler registered — return error (does not crash the unit)
return f"bgRFC: no handler registered for {func_name!r}"
_func_desc, handler = entry
# For bgRFC buffered calls, params are not yet deserialized (OG-06-02).
# Pass an empty request dict until live-capture confirms the encoding.
# Say so whenever the call carries more than the name we consumed: the
# handler is about to run against no data, and doing that to a business
# handler without a word is worse than the missing feature itself.
consumed = len(call_bytes) if nul_pos < 0 else nul_pos + 2
if len(call_bytes) > consumed:
_logger.warning(
"bgRFC: calling %s with an empty request — the buffered-call parameter "
"encoding is not implemented (OG-06-02), so %d byte(s) of this call are "
"being dropped",
func_name,
len(call_bytes) - consumed,
)
try:
handler({})
except Exception as exc: # noqa: BLE001 — isolate (T-06-U03)
return str(exc).splitlines()[0][:512]
return None
@staticmethod
def _extract_raw_param_from_frame(tlv: bytes, param_name: str) -> bytes | None:
"""Extract a raw (bytes) named param value from a TLV frame.
Returns the raw bytes value of the first param whose name matches
``param_name`` (case-insensitive). Returns None if not found.
Used for BGRFC_CALL_N entries (binary payload, not UTF-16LE strings).
"""
key = param_name.upper()
pos = 0
n = len(tlv)
current_name: str | None = None
while pos + 4 <= n:
tag = struct.unpack_from(">H", tlv, pos)[0]
length = struct.unpack_from(">H", tlv, pos + 2)[0]
pos += 4
if tag == _TAG_TERMINATOR:
break
if length == 0xFFFF:
if pos + 4 > n:
break
ext_len = struct.unpack_from(">I", tlv, pos)[0]
pos += 4
end = pos + ext_len
if end > n:
break
value = tlv[pos:end]
pos = end
else:
end = pos + length
if end > n:
break
value = tlv[pos:end]
pos = end
# Skip close tag
if pos + 2 <= n and struct.unpack_from(">H", tlv, pos)[0] == tag:
pos += 2
if tag == _TAG_PARAM_NAME:
current_name = _decode_utf16le(value)
elif tag == _TAG_PARAM_VALUE and current_name is not None:
if current_name.upper() == key:
return value
current_name = None
return None
# --------------------------------------------------------------------- #
# Request deserialize (SERVER-03)
# --------------------------------------------------------------------- #
@staticmethod
def _read_func_name(tlv: bytes) -> str:
"""Read the function-module name from tag 0x0102 (UTF-16LE).
Uses the bounds-checked invoke walker (T-05-C04); returns "" if absent.
"""
tags = _parse_tlv_stream(tlv)
raw = tags.get(_TAG_FUNC_NAME)
if raw is None:
return ""
return _decode_utf16le(raw)
@staticmethod
def _deserialize_request(tlv: bytes, func_desc: FunctionDesc | None) -> dict[str, Any]:
"""Walk 0x0201/0x0203 pairs and decode each into a typed Python value.
The request carries the client's IMPORTING values as-is (Pitfall 4); each
is decoded via ``codec.decode(field.rfctype, raw, field)`` per the
registered FunctionDesc. Unknown param names are ignored defensively.
When ``func_desc`` is ``None`` (transactional handler registered without a
descriptor, or dedup short-circuit caller), returns an empty dict — no
deserialization is attempted.
Registered-server inbound path: SAP encodes params in a 0x5001 compact
block (no 0x0201/0x0203 pairs). When the primary walk finds nothing, fall
back to ``_extract_5001_params`` which decodes the compact ASCII format.
"""
if func_desc is None:
return {}
name_to_field: dict[str, FieldDesc] = {f.name.upper(): f for f in func_desc.parameters}
request: dict[str, object] = {}
for name, raw in _extract_name_value_pairs(tlv):
field = name_to_field.get(name.upper())
if field is None:
continue
request[field.name] = decode(field.rfctype, raw, field)
# Registered-server inbound: 0x5001 compact param block (no 0x0201/0x0203)
if not request:
tags = _parse_tlv_stream(tlv)
raw_5001 = tags.get(0x5001)
if raw_5001 is not None:
for name, value_str in _extract_5001_params(raw_5001).items():
field = name_to_field.get(name.upper())
if field is not None:
request[field.name] = value_str
return request
# --------------------------------------------------------------------- #
# Transactional dispatch helpers (TRFC-03)
# --------------------------------------------------------------------- #
@staticmethod
def _extract_tid_from_frame(tlv: bytes) -> str:
"""Extract the ARFCTID parameter value from an ARFC_DEST_SHIP frame.
Tries 0x0201/0x0203 pairs first (offline fixtures / Python-built frames),
then falls back to the 0x5001 compact block (live SAP inbound tRFC frames
use NgRfc format — same encoding as Phase 5 registered-server inbound).
Returns the TID string (stripped of padding), or ``""`` if absent.
"""
for name, val in _extract_name_value_pairs(tlv):
if name.upper() == _PARAM_ARFCTID:
return _decode_utf16le(val)
# Fallback: live SAP sends params in 0x5001 compact block (ARFC_DEST_SHIP)
raw_5001 = _parse_tlv_stream(tlv).get(0x5001)
if raw_5001 is not None:
tid, _ = _extract_trfc_params_from_5001(raw_5001)
return tid
return ""
@staticmethod
def _extract_param_from_frame(tlv: bytes, param_name: str) -> str:
"""Extract any named UTF-16LE CHAR param value from a TLV frame.
Tries 0x0201/0x0203 pairs first (offline fixtures / Python-built frames),
then falls back to the 0x5001 compact block (live SAP inbound tRFC frames).
Used for ARFCFNAM (the wrapped function module name) and other metadata
params in the ARFC_DEST_SHIP frame. Returns ``""`` if absent.
"""
key = param_name.upper()
for name, val in _extract_name_value_pairs(tlv):
if name.upper() == key:
return _decode_utf16le(val)
# Fallback: live SAP sends params in 0x5001 compact block (ARFC_DEST_SHIP)
raw_5001 = _parse_tlv_stream(tlv).get(0x5001)
if raw_5001 is not None:
tid, arfcfnam = _extract_trfc_params_from_5001(raw_5001)
if key == "ARFCFNAM":
return arfcfnam
if key == _PARAM_ARFCTID:
return tid
return ""
def _build_rfc_executed_response(self) -> bytes:
"""Build the RFC_EXECUTED wire response (SDK type definitions, value 0x10 = 16).
The response return code is _RC_EXECUTED (16). SAP's client interprets
this as "TID already executed" and does NOT raise an error; it is a
normal flow indicator for exactly-once dedup (TRFC-03).
Format mirrors _build_system_failure but uses _RC_EXECUTED instead of
_RC_SYSTEM_FAILURE. No error-message TLV is emitted (not an error path).
"""
return b"".join(
[
tlv_record(_TAG_RESPONSE_START),
tlv_record(_TAG_RETURN_CODE, struct.pack(">I", _RC_EXECUTED)),
tlv_record(_TAG_TERMINATOR),
]
)
def _build_rfc_ok_response(self) -> bytes:
"""Build a minimal RFC_OK (return-code 0) response with no output params.
Used for tRFC success: ARFC_DEST_SHIP has no EXPORTING params (tRFC design).
Format: 0x0500 empty + 0x0420 = 0 + 0xFFFF.
"""
return b"".join(
[
tlv_record(_TAG_RESPONSE_START),
tlv_record(_TAG_RETURN_CODE, struct.pack(">I", _RC_OK)),
tlv_record(_TAG_TERMINATOR),
]
)
# --------------------------------------------------------------------- #
# Response serialize (SERVER-04) — mirror build_invoke_request, flipped
# --------------------------------------------------------------------- #
def _build_response(self, func_desc: FunctionDesc | None, result: dict[str, Any]) -> bytes:
"""Serialize a handler return dict to the response TLV stream (SERVER-04).
Mirror of ``invoke.build_invoke_request`` with directions flipped: emit the
0x0500 response-start marker, the 0x0420 return code (0 = success), then one
record group per EXPORTING/CHANGING/TABLE param the handler returned, then
the 0xFFFF terminator. Reuses ``tlv_record`` + ``codec.encode`` — NO second
TLV writer (RESEARCH Anti-Pattern).
Scalars and structures use the 0x0201(name)/0x0203(value) pair. A TABLE
parameter must NOT: it needs the table protocol, exactly as the client side
does. Emitting a table as a scalar 0x0203 value is the server-direction twin
of the mistyping that made client calls fail with
CALL_FUNCTION_ILLEGAL_P_TYPE.
Server-direction table shape, from the golden captures of a real SAP server
(tests/golden/framing/rfc_read_table_response.bin, and the compressed
metadata response): 0x0301(name) 0x0330(dm id) 0x0302(row_size,row_count)
then one 0x0304 per row. No 0x0306 end tag — in that capture each table runs
straight into the next 0x0301.
When ``func_desc`` is ``None`` (handler registered without a descriptor),
only the success header and terminator are emitted — no output params.
"""
result_upper = {k.upper(): v for k, v in result.items()}
dm_ids: list[str] = [] # DM table ids run from 1 in emission order
parts: list[bytes] = [
tlv_record(_TAG_RESPONSE_START),
tlv_record(_TAG_RETURN_CODE, struct.pack(">I", _RC_OK)),
]
if func_desc is not None:
for field in func_desc.parameters:
if field.direction not in _RESPONSE_DIRECTIONS:
continue # pure IMPORTING — client sent it, server does not echo
name_upper = field.name.upper()
if name_upper not in result_upper:
# Skipping is right — output params are optional — but a handler
# that meant to fill this one gets no hint: the client just sees
# the key missing from its result dict.
_logger.debug(
"server: handler for %s returned no value for output parameter "
"%s; it will be absent from the client's result",
func_desc.name,
field.name,
)
continue
value = result_upper[name_upper]
if field.rfctype == RFCTYPE_TABLE:
parts.extend(self._build_table_records(field, value, len(dm_ids) + 1))
dm_ids.append(field.name)
continue
encoded = encode(field.rfctype, value, field)
parts.append(tlv_record(_TAG_PARAM_NAME, field.name.encode("utf-16-le")))
parts.append(tlv_record(_TAG_PARAM_VALUE, encoded))
parts.append(tlv_record(_TAG_TERMINATOR))
return b"".join(parts)
@staticmethod
def _build_table_records(field: FieldDesc, rows: Any, dm_id: int) -> list[bytes]:
"""Serialize one TABLE output parameter using the table protocol.
An empty table is declared by name only, matching the client side where an
empty table needs no data block.
"""
if field.type_desc is None:
raise ValueError(
f"cannot encode TABLE parameter {field.name!r}: its row layout is "
f"missing (type_desc is None)"
)
row_list = list(rows) if rows else []
parts = [tlv_record(_TAG_TABLE_NAME, field.name.encode("utf-16-le"))]
if not row_list:
return parts
row_size = field.type_desc.uc_size if field.unicode_mode else field.type_desc.nuc_size
all_rows = encode(RFCTYPE_TABLE, row_list, field)
parts.append(tlv_record(_TAG_DM_TABLE_ID, struct.pack(">I", dm_id)))
parts.append(tlv_record(_TAG_TABLE_INFO, struct.pack(">II", row_size, len(row_list))))
for i in range(len(row_list)):
parts.append(tlv_record(_TAG_TABLE_ROW, all_rows[i * row_size : (i + 1) * row_size]))
return parts
def _build_system_failure(self, message: str) -> bytes:
"""Serialize an RFC SYSTEM_FAILURE response (D-03 / T-05-C02).
Non-zero return code (0x0420) + an error-message TLV (0x0402, UTF-16LE).
``message`` is sanitized: it is the caller-supplied ``str(exc)`` only — no
full traceback and no inbound credentials are ever placed here.
"""
safe = self._sanitize_message(message)
return b"".join(
[
tlv_record(_TAG_RESPONSE_START),
tlv_record(_TAG_RETURN_CODE, struct.pack(">I", _RC_SYSTEM_FAILURE)),
tlv_record(_TAG_ERROR_MESSAGE, safe.encode("utf-16-le")),
tlv_record(_TAG_TERMINATOR),
]
)
@staticmethod
def _sanitize_message(message: str) -> str:
"""Collapse a failure message to a single line (no traceback leakage).
Only the first line is kept and length-bounded, so a handler that raises
with an embedded traceback or a multi-line dump cannot leak it onto the
wire (T-05-C02). Credentials never reach this path (they live only in the
auth TLV, never in str(exc)).
"""
if not message:
return ""
first_line = message.splitlines()[0]
return first_line[:512]
# --------------------------------------------------------------------- #
# Authentication (SERVER-05) — placeholder until Task 2 wires the callback
# --------------------------------------------------------------------- #
def _check_auth(self, tlv: bytes) -> bool:
"""Extract inbound credentials and consult the auth callback (SERVER-05).
The user (tag 0x0111) and password (tag 0x0117) are read from the inbound
credential TLV; the secret is unscrambled with the symmetric
``_ab_scramble`` (its own inverse). The callback is invoked as
``fn(user=..., password=...)``. Neither value is ever written to a log or
echoed (T-04-CRED / T-05-C03). Returns True when no callback is set
(allow-all) or the callback returns truthy.
"""
if self._auth_check is None:
return True
user, password = self._extract_credentials(tlv)
try:
return bool(self._auth_check(user=user, password=password))
except TypeError:
# Tolerate a positional single-arg callback: fn(user).
return bool(self._auth_check(user))
@staticmethod
def _extract_credentials(tlv: bytes) -> tuple[str | None, str | None]:
"""Read user (0x0111) + unscrambled password (0x0117) from inbound TLV.
Returns ``(user, password)``; either may be ``None`` if the field is
absent (registration-mode inbound calls may pre-authenticate — A3). The
password 0x0117 value is ``seed(4B LE) + scramble(passwd, seed)``;
``_ab_scramble`` is symmetric so the same routine recovers the plaintext.
The plaintext is returned to the callback ONLY — never logged (T-05-C03).
"""
tags = _parse_tlv_stream(tlv)
user_raw = tags.get(_TAG_USER)
user = _decode_utf16le(user_raw) if user_raw else None
pwd_raw = tags.get(_TAG_PASSWORD)
password: str | None = None
if pwd_raw and len(pwd_raw) >= 4:
seed = struct.unpack_from("<I", pwd_raw, 0)[0]
body = bytearray(pwd_raw[4:])
_ab_scramble(body, seed)
password = bytes(body).decode("latin-1", "replace")
return user, password
# --------------------------------------------------------------------- #
# Blocking serve loop + sync facade (SERVER-06, D-01/D-02) — Task 2
# --------------------------------------------------------------------- #
def _serve_blocking(self) -> None: # pragma: no cover - live path
"""Blocking serve loop: register with GW, signal ready, dispatch inbound calls.
SDK-verified protocol (sdk_reg.pcap + sdk_listen.pcap ground truth):
1. NI init (64B, type=0x020b) → recv NI response (discard)
2. 512B hostname+TPNAME frame → SMGW: "Registered Server"
3. 06d1 REG_WAITING (80B) → SMGW: "Waiting for CPI-C Call"
4. recv 06cf (125B) — GW assigns session handle at [40:48]
5. recv loop: 0603 = inbound call → dispatch → 0608 response + cleanup
06cf = GW re-handle (after re-register) → update handle
"""
import socket as _socket
from saprfclib.transport import connect_tcp
program_id = self._params["program_id"]
gwhost = self._params.get("gwhost", "localhost")
gwserv = self._params["gwserv"]
transport = connect_tcp(gwhost, _gwserv_port(gwserv))
self._transport = transport
try:
local_ip_str = transport.local_address[0]
local_ip_bytes = _socket.inet_aton(local_ip_str)
try:
local_host = _socket.gethostname()
except Exception:
local_host = "saprfclib"
prog_id_enc = program_id.encode("ascii")
# --- NI init (64B, SDK-verified from sdk_reg.pcap PKT3) ---
_pid9 = prog_id_enc[:9]
proc_name_10 = _pid9 + b"\x00" + b"\x20" * (9 - len(_pid9))
local_host_8 = local_host[:8].encode("ascii", "replace").ljust(8, b"\x20")
# 16-byte field, so truncate at 16 — not 8.
#
# This read prog_id_enc[:8] and was invisible for the capture it was
# written from: that used the program ID "python3", seven characters,
# so slicing at 8 changed nothing and ljust(16) produced the right
# bytes. Any longer ID was silently cut in half — "SAPRFC_TEST"
# registered as "SAPRFC_T" — and the gateway then never matches the
# SM59 destination, so the server waits for calls that never arrive
# with nothing anywhere reporting a problem.
prog_id_16 = prog_id_enc[:16].ljust(16, b"\x20")
ni_init = (
b"\x02\x0b"
+ local_ip_bytes
+ b"\x00\x00\x00\x00"
+ proc_name_10
+ b"1100"
+ b"\x00\x00\x00\x00"
+ b"\x00\x06"
+ local_host_8
+ prog_id_16
+ b"\x06\xcb\xff\xff"
+ b"\x00" * 6
)
assert len(ni_init) == 64
transport.send_message(ni_init)
transport.recv_message() # NI response (sdk_reg.pcap PKT5) — discard
# --- 512B hostname+TPNAME frame (sdk_reg.pcap PKT7) ---
frame_512 = bytearray(b"\x20" * 512)
_lh = local_host.encode("ascii", "replace")[:127]
frame_512[: len(_lh)] = _lh
frame_512[len(_lh)] = 0
_pid = prog_id_enc[:63]
frame_512[128 : 128 + len(_pid)] = _pid
frame_512[128 + len(_pid)] = 0
transport.send_message(bytes(frame_512))
# --- 06d1 REG_WAITING (sdk_listen.pcap PKT4) ---
# RfcListenAndDispatch sends this to signal CMACCP "I'm ready to accept".
# GW transitions our entry to "Waiting for CPI-C Call" in SMGW.
transport.send_message(_build_reg_waiting())
# --- recv 06cf/06ce — GW assigns session handle ---
# sdk_listen.pcap PKT5: type 0x06CF when a call arrives immediately.
# Empirically observed: type 0x06CE is also sent by GW (queued-call path).
# Both carry the GW session handle at bytes [40:48].
cf_resp = transport.recv_message()
cf_ft = int.from_bytes(cf_resp[0:2], "big") if len(cf_resp) >= 2 else 0
gw_handle: bytes = cf_resp[40:48] if len(cf_resp) >= 48 else b" "
_logger.debug(
"[saprfclib-server] LISTENING — program_id=%r gw_frame=0x%04x gw_handle=%r",
program_id,
cf_ft,
gw_handle,
)
while not self._stopped:
try:
frame = transport.recv_message()
except (EOFError, OSError) as _e:
_logger.debug("[saprfclib-server] loop exit: %s: %s", type(_e).__name__, _e)
break
if len(frame) < 2:
continue
ft = int.from_bytes(frame[0:2], "big")
_logger.debug("[saprfclib-server] frame type=0x%04x len=%d", ft, len(frame))
if ft in (0x06CF, 0x06CE):
# GW (re-)assigned session handle — 06CF after 06d0 re-register,
# 06CE for queued-call dispatch.
gw_handle = frame[40:48] if len(frame) >= 48 else gw_handle
_logger.debug("[saprfclib-server] gw handle=%r", gw_handle)
elif ft == 0x0603:
self._dispatch_and_reply_sync(transport, frame, gw_handle)
else:
_logger.debug("[saprfclib-server] unhandled frame 0x%04x, skip", ft)
_logger.debug("[saprfclib-server] serve loop exited")
finally:
self._transport = None
transport.close()
def _dispatch_and_reply_sync( # pragma: no cover
self, transport: Transport, frame: bytes, gw_handle: bytes
) -> None:
"""Dispatch one 0603 inbound call and send the 0608 response + cleanup frames.
sdk_listen.pcap PKT8 (0608 response): 80B GW header with GW handle at [40:48]
+ RFC TLV from dispatch_inbound. Post-call: PKT9 (060b) + PKT10 (06d2) +
PKT11 (06d0) to signal GW we processed the call and are ready for the next.
"""
try:
response_tlv = self.dispatch_inbound(frame)
full_response = _wrap_in_0608(response_tlv, gw_handle)
transport.send_message(full_response)
_logger.debug(
"[saprfclib-server] dispatch OK — %dB TLV wrapped in 0608 (%dB)",
len(response_tlv),
len(full_response),
)
# Post-call cleanup (sdk_listen.pcap PKT9/PKT10/PKT11)
transport.send_message(_build_post_call_b(gw_handle)) # 060b
transport.send_message(_build_post_call_d2(gw_handle)) # 06d2
transport.send_message(_build_re_reg()) # 06d0 re-register
except Exception as e: # noqa: BLE001
_logger.error("[saprfclib-server] dispatch ERROR: %s: %s", type(e).__name__, e)
def serve_forever(self) -> None: # pragma: no cover - live path, exercised in Plan 04
"""Blocking facade: run the serve loop in a daemon thread (D-02).
The caller needs no asyncio knowledge. Call stop()/close() from another
thread to tear down cleanly (Pitfall 5).
"""
if self._thread is not None and self._thread.is_alive():
raise RuntimeError("server already running")
self._stopped = False
_exc: list[BaseException] = []
def _run() -> None:
try:
self._serve_blocking()
except Exception as e: # noqa: BLE001
_logger.error("[saprfclib-server] FATAL: %s: %s", type(e).__name__, e)
_exc.append(e)
self._thread = threading.Thread(target=_run, name="saprfclib-server", daemon=True)
self._thread.start()
self._thread.join()
if _exc:
raise _exc[0]
def stop(self) -> None: # pragma: no cover - live path
"""Signal the serve loop to stop: set the stopped flag and close the socket.
Closing the socket unblocks the blocking recv_message() call in the serve
loop so it exits cleanly. Safe to call from any thread (Pitfall 5).
"""
self._stopped = True
transport = getattr(self, "_transport", None)
if transport is not None:
try:
transport.close()
except Exception: # noqa: BLE001
pass
def close(self) -> None: # pragma: no cover - live path
"""Tear the server down: stop the loop and join the background thread."""
self.stop()
thread = self._thread
if thread is not None and thread.is_alive():
thread.join(timeout=5.0)
self._thread = None