diff --git a/README.md b/README.md index 7ccea76..b519498 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Capa is a small, capability-typed programming language. Every function declares the authorities it holds (`Fs`, `Net`, `Stdio`, -`Clock`, `Random`, `Env`, `Db`, `Proc`, `Unsafe`), the analyzer +`Clock`, `Random`, `Env`, `Db`, `Proc`, `Serve`, `Unsafe`), the analyzer enforces those declarations statically, and the compiler emits **CycloneDX SBOM**, **SPDX 2.3**, **VEX**, and **SLSA Build L1 provenance** documents directly from the same capability signatures. @@ -194,8 +194,10 @@ import is needed at the Component Model boundary. ## Standard library + seed libraries The runtime ships built-in types (`Result`, `Option`, `List`, -`Map`, `Set`, `JsonValue`) and built-in capabilities (`Stdio`, -`Fs`, `Net`, `Env`, `Clock`, `Random`, `Db`, `Proc`, `Unsafe`). +`Map`, `Set`, `JsonValue`) and ten built-in capabilities (`Stdio`, +`Fs`, `Net`, `Env`, `Clock`, `Random`, `Db`, `Proc`, `Serve`, +`Unsafe`). `Serve` (inbound TCP) and `Unsafe` are Python-backend +only; `capa --wasm` rejects a program whose signatures reach either. Full reference in [`docs/stdlib.md`](docs/stdlib.md). Eight **seed libraries** live in standalone repos and are @@ -289,9 +291,10 @@ backend (with a Python/Wasm output parity harness), and Hypothesis-based property tests. The transpiler suite actually executes the generated Python and checks stdout; the property suite fuzzes the full pipeline with arbitrary text and -syntax-aware Capa programs. The Wasm backend runs every capability -(Fs, Env, Clock, Stdio, Net, Random, Db, Proc) and the full language -surface with output byte-identical to the Python reference, and +syntax-aware Capa programs. The Wasm backend runs every capability it +supports (Fs, Env, Clock, Stdio, Net, Random, Db, Proc, all but the +Python-only Serve and Unsafe, which it rejects loudly) and the full +language surface with output byte-identical to the Python reference, and cross-function capability attenuation is enforced soundly at the Wasm runtime via host-side handle tables. diff --git a/capa/analyzer/_declarations.py b/capa/analyzer/_declarations.py index 60b3de4..ec472cf 100644 --- a/capa/analyzer/_declarations.py +++ b/capa/analyzer/_declarations.py @@ -503,8 +503,9 @@ def _ty_contains_fun_deep( exact quarantine violation F2 relies on being impossible), so it must be rejected wherever it is reachable, not only as a literal head. The walk mirrors the Wasm backend's - ``_unsafe_bearing_type_names`` fixpoint (struct fields + - variant payloads + generic args) and is cycle-guarded on the + ``_python_only_caps._cap_bearing_type_names`` fixpoint + (struct fields + variant payloads + generic args) and is + cycle-guarded on the type name for recursive types (``type Tree { kids: List }``).""" from . import SymbolKind diff --git a/capa/analyzer/_ifc.py b/capa/analyzer/_ifc.py index db91157..0712e8e 100644 --- a/capa/analyzer/_ifc.py +++ b/capa/analyzer/_ifc.py @@ -62,6 +62,20 @@ ("Fs", "write"): {0, 1}, ("Db", "exec"): {0, 1}, ("Db", "query"): {0, 1}, + # Serve.send writes bytes to whoever is on the other end of an + # inbound connection -- exfiltration exactly like Net.post. Only + # argument 1 (the payload) is a sink; argument 0 is the connection + # id the runtime handed out, not program data, so gating it would + # be noise. + # + # It is spelled ``send`` and not ``write`` because the summary pass + # in ``_ifc_summary`` attributes a sink to a capability BY METHOD + # NAME (it has no receiver type at that point), which is sound only + # while each sink method name belongs to exactly one capability. + # ``Fs.write`` already owns "write", so a ``Serve.write`` made every + # ``fs.write`` report Serve as a reached capability too -- caught by + # tests/test_unaudited_secret_sink_fact.py when this landed. + ("Serve", "send"): {1}, } # Built-in capability methods that PRODUCE secret data -- the sources. @@ -78,6 +92,26 @@ # and over-labelling it would warn on every legitimate file echo. A # program that does hold a secret in a file can annotate the binding # ``@secret`` explicitly. Future levels could make this configurable. +# +# ``Serve.read`` is deliberately NOT a source either, and this is the +# one entry whose ABSENCE is a decision worth spelling out. Serve +# (2026-07) is the language's first INBOUND data source, so it is the +# first time the question "is data arriving from outside secret?" has +# an answer to give. It is ``@public``. +# +# The reason is that this lattice models CONFIDENTIALITY -- who is +# allowed to LEARN a value -- and not integrity or taint. An inbound +# request is untrusted, but "untrusted" is an integrity property, and +# labelling it ``@secret`` would encode it in the wrong lattice: the +# immediate consequence is that echoing a request back to the client +# that sent it (the single most ordinary thing a server does) becomes +# a reported violation. The useful signal would drown in that noise. +# +# ``Serve.read`` being ``@public`` therefore asserts only "these bytes +# are not a secret whose disclosure this analysis must prevent". It +# asserts NOTHING about whether they can be trusted. Integrity / +# taint tracking would be a second lattice, not a relabelling of this +# one. _SECRET_SOURCES: frozenset = frozenset({ ("Env", "get"), }) diff --git a/capa/analyzer/_ifc_summary.py b/capa/analyzer/_ifc_summary.py index 8a9e950..80ecdb5 100644 --- a/capa/analyzer/_ifc_summary.py +++ b/capa/analyzer/_ifc_summary.py @@ -1028,7 +1028,12 @@ def _taint_of_method_call( # this argument reach this built-in sink, so record its # capability against EACH of them. A sink method name is # unique to one capability in ``_PUBLIC_SINKS``, so - # ``_cap`` is determinate. + # ``_cap`` is determinate. That uniqueness is not a + # happy accident to be relied on quietly: it is now + # pinned by + # ``test_sink_method_names_are_unique_per_capability`` + # in tests/test_serve_capability.py, which fired when + # Serve was first given a ``write``. self._attribute_sink_caps(arg_srcs[pos], (_cap,)) # A mutating container method (push / add / set) routes the diff --git a/capa/builtins.py b/capa/builtins.py index 829bab6..65a274c 100644 --- a/capa/builtins.py +++ b/capa/builtins.py @@ -302,6 +302,41 @@ def fun(*params_then_ret: Ty) -> TyFun: ("allows", fun(TyString, TyBool), []), ("exec", fun(TyString, TyString, _str_res), []), ], + "Serve": [ + # Inbound TCP listener (2026-07). Connection-level on purpose: + # the trusted runtime binds, accepts, and moves bytes; HTTP (or + # any other protocol) is parsed by ordinary Capa code, so a + # protocol bug is not a bug in the trusted computing base. + # + # Attenuation domain is the (bind address, port) pair, spelled + # ``"addr:port"`` / ``"addr:lo-hi"`` / ``"addr:*"`` with ``*`` + # also accepted as the address. ``allows`` mirrors ``listen``'s + # argument shape rather than the one-String shape of the other + # caps, precisely so "allows(a, p)" answers "would listen(a, p) + # be permitted". + # + # Bytes are ``List`` with each element in ``0..=255``. + # ``recv`` returning an EMPTY list means EOF (the peer closed). + # Spelled recv/send, the standard socket verbs; ``write`` in + # particular is unavailable because IFC sink attribution + # requires sink method names to be unique per capability and + # ``Fs.write`` already holds that name. + # Serve is sequential: one open connection at a time, no + # threads and no async. ``accept`` and ``read`` are bounded by + # a timeout and return ``Err`` rather than hanging. + # + # Python backend only: the Wasm emitter rejects any signature + # reaching Serve (no vendored ``wasi:sockets``). + ("restrict_to", fun(TyString, TyName("Serve")), []), + ("allows", fun(TyString, TyInt, TyBool), []), + ("listen", fun(TyString, TyInt, _unit_res), []), + ("local_port", fun(res(TyInt, _io_err)), []), + ("accept", fun(res(TyInt, _io_err)), []), + ("recv", fun(TyInt, TyInt, res(lst(TyInt), _io_err)), []), + ("send", fun(TyInt, lst(TyInt), _unit_res), []), + ("close", fun(TyInt, _unit_res), []), + ("stop", fun(_unit_res), []), + ], "JsonValue": [ ("is_null", fun(TyBool), []), ("as_bool", fun(opt(TyBool)), []), diff --git a/capa/ir/_capa_types.py b/capa/ir/_capa_types.py index 68e0c7b..7639512 100644 --- a/capa/ir/_capa_types.py +++ b/capa/ir/_capa_types.py @@ -31,7 +31,32 @@ # that module skips wholesale when Hypothesis is absent and a # registry guard must never be skippable.) BUILTIN_CAPS: frozenset[str] = frozenset({ - "Stdio", "Fs", "Net", "Env", "Clock", "Random", "Proc", "Db", "Unsafe", + "Stdio", "Fs", "Net", "Env", "Clock", "Random", "Proc", "Db", + "Serve", "Unsafe", +}) + + +# The built-in caps the Wasm backends cannot implement at all. A +# program whose signatures reach one of these is rejected up front +# with a diagnostic naming the offending sites, so it never reaches +# lowering or the host bridges. +# +# TWO independent paths must reject, because each is reachable on +# its own: the Wasm emitter's discovery pass (``capa --wasm``) and +# WIT generation (``capa --wit``, which never runs the emitter). The +# shared reachability scan they both call lives in +# [`_python_only_caps.py`](_python_only_caps.py). +# +# These are PERMANENT stances, not backlog items: +# +# - ``Unsafe`` grants raw pointer / FFI / mmap-class primitives that +# have no sandboxed Wasm equivalent (slice 7 / D5, 2026-05). +# - ``Serve`` needs ``wasi:sockets`` to bind a listening socket. That +# world is not vendored in ``capa/wasi_wit`` and is not reachable +# from the wasmtime-py 44 bindings the hosts are built on, so there +# is no inbound socket for the guest to be handed (2026-07). +PYTHON_ONLY_CAPS: frozenset[str] = frozenset({ + "Unsafe", "Serve", }) @@ -60,6 +85,19 @@ # derived from ``BUILTIN_CAPS - HANDLE_BEARING_CAPS`` so that adding # a capability forces an explicit choice instead of silently # defaulting to "erased". +# +# ``Unsafe`` and ``Serve`` are here because they are members of +# ``PYTHON_ONLY_CAPS``: the Wasm backends reject them before lowering, +# so neither ever reaches a Wasm slot and "handle-bearing" would be a +# claim about codegen that never runs. Erased is also the SAFER +# classification for a rejected cap. If the rejection ever developed a +# hole, an erased cap pushes no value and the module fails to link +# against a host import nobody defined (loud); a handle-bearing cap +# would occupy an i32 slot with no entry in the host's root-handle +# map and silently fall back to the ``Fs`` root, which is exactly the +# quiet cross-cap substitution +# ``test_host_root_handle_map_covers_every_handle_bearing_cap`` +# exists to prevent. ERASED_CAPS: frozenset[str] = frozenset({ - "Stdio", "Random", "Unsafe", + "Stdio", "Random", "Unsafe", "Serve", }) diff --git a/capa/ir/_emit_wasm/_discovery.py b/capa/ir/_emit_wasm/_discovery.py index 3b3766d..28877b6 100644 --- a/capa/ir/_emit_wasm/_discovery.py +++ b/capa/ir/_emit_wasm/_discovery.py @@ -31,7 +31,6 @@ from __future__ import annotations -import re from .._nodes import ( Module, Instr, Value, Function, @@ -41,6 +40,7 @@ ) from .._lower_pattern import PatStruct, PatOr from .._capa_types import BUILTIN_CAPS +from .._python_only_caps import find_rejection from .._emit_wit import _WIT_SIGNATURES from .._walk import iter_functions, walk_instrs, walk_module from ._layout import ( @@ -49,6 +49,7 @@ ) + def _pattern_has_str_literal(pat) -> bool: """True iff ``pat`` carries a String literal anywhere in its sub-pattern tree: a variant payload (``Ok("yes")`` / nested @@ -469,107 +470,37 @@ def _discover(self, module: Module) -> None: which read as "this is a backlog item" rather than the permanent stance it actually is. The single early raise is more honest and points the user at the correct - workaround (Python backend or refactor the call site).""" - self._reject_unsafe_signatures(module) + workaround (Python backend or refactor the call site). + + 2026-07: the same scan now covers every member of + ``PYTHON_ONLY_CAPS``, so ``Serve`` gets the identical early, + site-listing rejection.""" + self._reject_python_only_cap_signatures(module) for fn in iter_functions(module): self._discover_instrs(fn.body) - @staticmethod - def _type_name_tokens(ty: str) -> set[str]: - """Every type-name identifier appearing in a CIR type - string, including generic args and tuple elements. A simple - word scan is sound for the Unsafe reachability check: the - only false positives would be a user type whose name happens - to be a substring of another, which the ``\\b`` boundaries - rule out.""" - if not ty: - return set() - return set(re.findall(r"[A-Za-z_]\w*", ty)) - - def _unsafe_bearing_type_names(self, module: Module) -> set[str]: - """Fixpoint set of struct / sum type names that - transitively reach ``Unsafe`` through a field or variant - payload, so a parameter of such a type is rejected even - when ``Unsafe`` never appears literally in the signature - (audit 2026-06-17 C5(b)).""" - field_tys: dict[str, list[str]] = {} - for decl in module.types: - fields = getattr(decl, "fields", None) - if fields is not None: - field_tys[decl.name] = [f.ty for f in fields] - continue - variants = getattr(decl, "variants", None) - if variants is not None: - field_tys[decl.name] = [ - pty for v in variants for pty in v.payload_tys - ] - bearing: set[str] = set() - changed = True - while changed: - changed = False - for name, tys in field_tys.items(): - if name in bearing: - continue - for ty in tys: - toks = self._type_name_tokens(ty) - if "Unsafe" in toks or toks & bearing: - bearing.add(name) - changed = True - break - return bearing - - def _reject_unsafe_signatures(self, module: Module) -> None: - """Surface ``Unsafe``-reaching parameters at discovery time - with a diagnostic that names the function, the parameter, - and the only two valid responses (run on the Python - backend, or remove the Unsafe argument). Scans both - top-level functions and impl methods. - - Audit 2026-06-17 C5(b): the check walks each param type - RECURSIVELY through named struct fields, sum-variant - payloads, and generic arguments, so a param of a type that - merely *contains* Unsafe (``type Wrapper { u: Unsafe }``) - is rejected loud rather than slipping through to emit an - invalid ``call $py_import``. Mirrors the manifest's - reachability traversal: Unsafe is detected wherever it is - reachable through the type, not only as a literal head.""" - unsafe_bearing = self._unsafe_bearing_type_names(module) - - def reaches_unsafe(ty: str) -> bool: - for name in self._type_name_tokens(ty): - if name == "Unsafe" or name in unsafe_bearing: - return True - return False - - offenders: list[str] = [] - for fn in module.functions: - for p in fn.params: - if reaches_unsafe(p.ty): - offenders.append(f"{fn.name}({p.name}: {p.ty})") - for impl in module.impls: - impl_label = ( - f"impl {impl.trait_name} for {impl.type_name}" - if impl.trait_name else f"impl {impl.type_name}" - ) - for method in impl.methods: - for p in method.params: - if reaches_unsafe(p.ty): - offenders.append( - f"{impl_label}::{method.name}" - f"({p.name}: {p.ty})" - ) - if not offenders: - return - sites = "\n - ".join(offenders) - raise WasmEmissionError( - "the Unsafe capability is intentionally not supported " - "on the Wasm backend (it grants raw pointer / FFI / " - "memory-map primitives that have no sandboxed Wasm " - "equivalent). Use the Python backend for these " - "functions, or refactor to remove the Unsafe " - "parameter.\n - " - f"{sites}" - ) + def _reject_python_only_cap_signatures(self, module: Module) -> None: + """Reject a program whose signatures reach a member of + ``PYTHON_ONLY_CAPS`` (``Unsafe``, ``Serve``), naming the + capability, why it can never work here, what to do instead, + and the offending sites. + + The scan itself lives in + [`_python_only_caps.py`](../_python_only_caps.py) because WIT + generation needs exactly the same predicate and is reachable + on its own (``capa --wit`` never runs this discovery pass). + Two copies of a security-relevant reachability check would + drift silently, so there is one. + + Raising early matters: pre-2026-05 the rejection happened deep + in cap-method dispatch ("capability method Unsafe.alloc has no + WIT/Wasm encoding yet; widen the signature tables"), which read + as a backlog item rather than the permanent stance it is. + """ + found = find_rejection(module) + if found is not None: + _cap, message = found + raise WasmEmissionError(message) def _discover_instrs(self, instrs: list[Instr]) -> None: for instr in walk_instrs(instrs): diff --git a/capa/ir/_emit_wit.py b/capa/ir/_emit_wit.py index 8cd7f55..29def9c 100644 --- a/capa/ir/_emit_wit.py +++ b/capa/ir/_emit_wit.py @@ -276,16 +276,36 @@ # generation time; the Wasm emitter mirrors this so the contract # stays in sync. # -# This is ``BUILTIN_CAPS`` MINUS ``Unsafe``, and the omission is -# deliberate, not drift: ``Unsafe`` is the Python-only FFI escape -# hatch and can never reach WIT generation. The Wasm emitter -# rejects an ``Unsafe`` anywhere in a signature up front (see -# ``_discovery._reject_unsafe_signatures``), so it never lands in -# ``used``. Adding it here would be actively wrong - the interface -# loop would then try to emit ``interface unsafe { ... }`` and -# raise ``UnsupportedCapabilityMethod`` on the first method +# This is ``BUILTIN_CAPS`` MINUS ``PYTHON_ONLY_CAPS`` (``Unsafe``, +# ``Serve``), and the omission is deliberate, not drift. +# +# WHAT ACTUALLY KEEPS THEM OUT (corrected 2026-07; the previous +# version of this comment described a mechanism that does not exist). +# It is NOT the Wasm emitter's discovery rejection: ``capa --wit`` is +# a standalone path that never runs the Wasm emitter at all. The two +# capabilities stay out of the emitted document for two DIFFERENT +# reasons, neither of which was what this comment used to claim: +# +# - ``Serve`` DOES land in ``used`` -- it has capability methods, so +# ``collect_used_capabilities`` records every ``serve.listen(...)`` +# call. It is dropped later, by the ``cap not in +# _KNOWN_CAPABILITIES`` guard in the interface loop of ``emit_wit``. +# - ``Unsafe`` never lands in ``used``, but because it is METHOD-LESS +# (no entry in ``capa.builtins.METHODS``), so there is no cap-method +# call for the collector to see. Its authority is exercised through +# the free functions ``py_import`` / ``py_invoke``. +# +# Both of those are SILENT drops, which is why ``emit_wit`` now raises +# ``PythonOnlyCapabilityInWit`` up front rather than relying on them: +# a document that quietly omits a capability the program genuinely +# holds misdescribes the program. This set remains the backstop for +# the interface loop. +# +# Adding a Python-only cap here would be actively wrong -- the +# interface loop would then try to emit ``interface serve { ... }`` +# and raise ``UnsupportedCapabilityMethod`` on the first method # instead of skipping the cap. Kept as a literal (rather than -# ``BUILTIN_CAPS - {"Unsafe"}``) so a newly added capability is +# ``BUILTIN_CAPS - PYTHON_ONLY_CAPS``) so a newly added capability is # NOT silently assumed to have WIT signatures; the guard in # ``TestCapabilityRegistry`` (tests/test_cap_handles.py) flags the # omission when a new cap appears. @@ -594,6 +614,29 @@ def __init__(self, cap: str, method: str): self.method = method +class PythonOnlyCapabilityInWit(Exception): + """Raised when WIT generation is asked to describe a program that + holds a ``PYTHON_ONLY_CAPS`` capability (``Unsafe``, ``Serve``). + + ``capa --wit`` is a STANDALONE path: it does not run the Wasm + emitter, so the discovery-time rejection never fires for it. Before + this existed, ``--wit`` on such a program exited 0 and printed a + document that silently omitted the capability -- and worse, whose + ``world`` block declared ``export main: func()`` while the real + ``main`` took the capability as a parameter. A document whose whole + purpose is to describe the program's interface to a host was + therefore describing a different program. + + Failing loud is the right call here rather than emitting a partial + document, because a program holding one of these capabilities can + never become a Wasm component at all: there is no host that could + consume the WIT, so nothing legitimate is lost.""" + + def __init__(self, cap: str, message: str): + super().__init__(message) + self.cap = cap + + def collect_used_capabilities(module: Module) -> dict[str, set[str]]: """Walk every instruction in every function and return a capability_name -> set-of-method-names mapping. @@ -692,6 +735,7 @@ def collect_used_capabilities(module: Module) -> dict[str, set[str]]: from ._capa_types import BUILTIN_CAPS, HANDLE_BEARING_CAPS +from ._python_only_caps import find_rejection as find_python_only_rejection def _module_calls_panic(module: Module) -> bool: @@ -785,7 +829,31 @@ def emit_wit( interfaces (resolved from the vendored WIT in ``capa/wasi_wit``) instead of the matching ``capa:host`` interfaces; every other capability keeps its ``capa:host`` interface (hybrid mode). See - ``docs/design/wasi_mode.md``.""" + ``docs/design/wasi_mode.md``. + + Raises ``PythonOnlyCapabilityInWit`` if ``module`` reaches a + ``PYTHON_ONLY_CAPS`` capability. ``--wit`` does not run the Wasm + emitter, so this is the only place that rejection can happen on + this path; without it the document would silently omit the + capability and misdescribe ``main``'s signature.""" + # Before ``used`` is even collected, so the check is independent of + # whether the capability happens to be method-bearing (``Serve`` + # is and shows up in ``used``; ``Unsafe`` is method-less and never + # does -- see the ``_KNOWN_CAPABILITIES`` comment). Both must be + # rejected, so this scans SIGNATURES, exactly as the Wasm + # discovery pass does, sharing that one implementation. + found = find_python_only_rejection( + module, + note=( + "There is no WIT to emit for it: a WIT document " + "describes a Wasm component, and this program cannot " + "be one." + ), + ) + if found is not None: + cap, message = found + raise PythonOnlyCapabilityInWit(cap, message) + used = collect_used_capabilities(module) if wasi: diff --git a/capa/ir/_python_only_caps.py b/capa/ir/_python_only_caps.py new file mode 100644 index 0000000..b77f2c8 --- /dev/null +++ b/capa/ir/_python_only_caps.py @@ -0,0 +1,175 @@ +"""Rejection of capabilities that only work on the Python backend. + +``PYTHON_ONLY_CAPS`` (see [`_capa_types.py`](_capa_types.py)) names the +built-in capabilities no Wasm backend can implement. Two independent +emit paths must refuse a program that reaches one: + +- the Wasm emitter's discovery pass + ([`_emit_wasm/_discovery.py`](_emit_wasm/_discovery.py)), and +- WIT generation ([`_emit_wit.py`](_emit_wit.py)), which is reachable + on its own via ``capa --wit`` WITHOUT going through the Wasm + emitter. + +The reachability scan lives here, once, rather than in either caller. +Duplicating it is exactly the failure mode ``_capa_types``' own +docstring warns about: two copies of a security-relevant predicate +drift, and the drift is silent. Keeping the reason strings next to the +scan also means a new Python-only capability is explained identically +on every path that rejects it. + +This module deliberately raises nothing itself. The two callers have +different exception types (``WasmEmissionError`` vs +``UnsupportedCapability``) and both are meaningful to their own users, +so the shared code produces the offender list and the message text and +lets each caller raise in its own vocabulary. +""" + +from __future__ import annotations + +import re + +from ._capa_types import PYTHON_ONLY_CAPS + + +# Why each ``PYTHON_ONLY_CAPS`` member cannot work on Wasm, phrased to +# sit inside "the X capability is intentionally not supported on the +# Wasm backend (...)". These are permanent stances, not backlog items, +# and the wording says so: someone who reads "not supported" should not +# go looking for the tracking issue. +PYTHON_ONLY_CAP_REASONS: dict[str, str] = { + "Unsafe": ( + "it grants raw pointer / FFI / memory-map primitives that " + "have no sandboxed Wasm equivalent" + ), + "Serve": ( + "binding a listening socket needs wasi:sockets, which is " + "neither vendored in capa/wasi_wit nor reachable from the " + "wasmtime-py bindings the Wasm hosts are built on, so a " + "guest can never be handed an inbound connection" + ), +} + + +def _type_name_tokens(ty: str) -> set[str]: + """Every type-name identifier appearing in a CIR type string, + including generic args and tuple elements. A simple word scan is + sound for the reachability check: the only false positives would be + a user type whose name happens to be a substring of another, which + the ``\\b`` boundaries rule out.""" + if not ty: + return set() + return set(re.findall(r"[A-Za-z_]\w*", ty)) + + +def _cap_bearing_type_names(module, cap: str) -> set[str]: + """Fixpoint set of struct / sum type names that transitively reach + ``cap`` through a field or variant payload, so a parameter of such + a type is caught even when ``cap`` never appears literally in the + signature (audit 2026-06-17 C5(b)).""" + field_tys: dict[str, list[str]] = {} + for decl in module.types: + fields = getattr(decl, "fields", None) + if fields is not None: + field_tys[decl.name] = [f.ty for f in fields] + continue + variants = getattr(decl, "variants", None) + if variants is not None: + field_tys[decl.name] = [ + pty for v in variants for pty in v.payload_tys + ] + bearing: set[str] = set() + changed = True + while changed: + changed = False + for name, tys in field_tys.items(): + if name in bearing: + continue + for ty in tys: + toks = _type_name_tokens(ty) + if cap in toks or toks & bearing: + bearing.add(name) + changed = True + break + return bearing + + +def find_offending_sites(module, cap: str) -> list[str]: + """Every parameter in ``module`` whose type reaches ``cap``, + rendered as a human-readable site label. + + Covers top-level functions and impl methods. The walk goes + RECURSIVELY through named struct fields, sum-variant payloads, and + generic arguments, so a parameter of a type that merely *contains* + the capability (``type Wrapper { u: Unsafe }``) is caught rather + than slipping through to emit an invalid call. + + Known limit: a capability reached ONLY through ``self`` inside an + impl method is not listed as its own site. Such a module is still + rejected, because constructing the receiver struct requires a + parameter of the capability's type somewhere, and THAT parameter is + listed. So the rejection is not missed; only the site list is less + complete than it could be. + """ + bearing = _cap_bearing_type_names(module, cap) + + def reaches(ty: str) -> bool: + for name in _type_name_tokens(ty): + if name == cap or name in bearing: + return True + return False + + sites: list[str] = [] + for fn in module.functions: + for p in fn.params: + if reaches(p.ty): + sites.append(f"{fn.name}({p.name}: {p.ty})") + for impl in module.impls: + impl_label = ( + f"impl {impl.trait_name} for {impl.type_name}" + if impl.trait_name else f"impl {impl.type_name}" + ) + for method in impl.methods: + for p in method.params: + if reaches(p.ty): + sites.append( + f"{impl_label}::{method.name}({p.name}: {p.ty})" + ) + return sites + + +def format_rejection(cap: str, sites: list[str], *, note: str = "") -> str: + """The diagnostic text for a rejected capability: what, why, what + to do instead, and where. + + The subject is always "the Wasm backend", on every path, because + WIT generation exists only to describe a Wasm component -- a + program that cannot run on Wasm has no meaningful WIT either. The + caller's own prefix (``capa: --wasm:`` / ``capa: --wit:``) already + tells the user which command they ran. ``note`` appends a + path-specific sentence where one genuinely adds information.""" + listed = "\n - ".join(sites) + extra = f" {note}" if note else "" + return ( + f"the {cap} capability is intentionally not supported " + f"on the Wasm backend ({PYTHON_ONLY_CAP_REASONS[cap]}). " + f"Use the Python backend for these functions, or " + f"refactor to remove the {cap} parameter.{extra}\n - " + f"{listed}" + ) + + +def find_rejection(module, *, note: str = ""): + """Return ``(cap, message)`` for the first Python-only capability + ``module`` reaches, or ``None`` if it reaches none. + + Per-capability rather than aggregated on purpose: the members have + entirely different reasons and entirely different "what to do + instead", so one raise naming one capability is more useful than a + combined message. Iteration is sorted for a deterministic + diagnostic when a program somehow reaches two. + """ + for cap in sorted(PYTHON_ONLY_CAPS): + sites = find_offending_sites(module, cap) + if sites: + return cap, format_rejection(cap, sites, note=note) + return None diff --git a/capa/runtime/__init__.py b/capa/runtime/__init__.py index ef4a54f..0851fe0 100644 --- a/capa/runtime/__init__.py +++ b/capa/runtime/__init__.py @@ -7,7 +7,7 @@ types used for fallible and optional values. - Concrete capabilities (see :mod:`._capabilities`): ``Stdio``, ``Fs``, ``Env``, ``Clock``, ``Random``, ``Net``, ``Proc``, ``Db``, - ``Unsafe``. Each is instantiated in ``main`` and threaded through + ``Serve``, ``Unsafe``. Each is instantiated in ``main`` and threaded through function arguments. There are no "ambient capabilities": importing this runtime does not grant any authority. - The Python-interop boundary (see :mod:`._pyinterop`): ``py_import`` @@ -45,6 +45,7 @@ Net, Proc, Random, + Serve, Stdio, Unsafe, _StubCapability, @@ -115,6 +116,7 @@ "Net", "Proc", "Db", + "Serve", "Unsafe", "py_import", "py_invoke", diff --git a/capa/runtime/_capabilities.py b/capa/runtime/_capabilities.py index e0eb803..c32843b 100644 --- a/capa/runtime/_capabilities.py +++ b/capa/runtime/_capabilities.py @@ -25,12 +25,15 @@ - ``Proc`` - sandboxed subprocess execution with basename-prefix attenuation (slice 15, 2026-05). - ``Net`` - HTTP GET with host-set attenuation. +- ``Serve`` - inbound TCP listener with bind-address + port-range + attenuation (2026-07, Python backend only). - ``Unsafe`` - the Python-interop trust boundary (method-less). """ from __future__ import annotations import os +import socket import sys import time from dataclasses import dataclass @@ -712,6 +715,449 @@ def post(self, url: str, body: str, max_bytes: Optional[int] = None): return Err(IoError("HTTP POST failed", str(e))) +# Bounded waits for the Serve capability. Nothing in Capa may block +# forever: ``Net.get`` bounds an outbound request at 10s and +# ``Proc.exec`` bounds a child at 30s, so the two inbound waits get +# the same treatment. ``accept`` waits for a client that may never +# arrive, which is closer to "wait for a child to finish" than to +# "wait for a server to answer", hence the 30s bound; a per-connection +# ``recv`` gets the same budget so a peer that opens a socket and then +# goes silent cannot pin the program. +# +# Both bounds return ``Err(IoError("... timed out", ...))``. They are +# NOT a way to poll: a program that wants to keep serving calls +# ``accept`` again after a timeout Err. +_SERVE_ACCEPT_TIMEOUT_SECS = 30.0 +_SERVE_CONN_TIMEOUT_SECS = 30.0 + +# Listen backlog. Serve is sequential by construction (one open +# connection at a time), so the kernel queue exists only to avoid +# refusing a client that arrives while the program is still writing +# the previous response. +_SERVE_BACKLOG = 8 + +# Hard ceiling on a single ``read``. The caller's ``max_bytes`` is +# clamped to this, so a program cannot ask the host to allocate an +# arbitrarily large buffer on behalf of a remote peer. A caller that +# wants more bytes loops. +_SERVE_MAX_READ = 1 << 20 + +# The rule a malformed ``restrict_to`` spec parses to: a host nothing +# matches and an empty port range. Attenuation cannot report an error +# (``restrict_to`` returns a ``Serve``, not a ``Result``), so a spec +# that does not parse must deny rather than be ignored - ignoring it +# would silently WIDEN authority, which is the one thing attenuation +# must never do. +_SERVE_DENY_ALL_RULE = ("", 1, 0) + + +class Serve: + """Capability to listen on a network address and accept inbound + connections, with first-class attenuation. + + This is the language's only INBOUND authority: ``Net`` reaches + out, ``Serve`` is reached. It is deliberately connection-level, + not HTTP-level - the trusted runtime binds a socket, hands out + accepted connections, and moves bytes. Request parsing (HTTP or + anything else) is ordinary Capa code in a library, so a protocol + bug is not a bug in the trusted computing base. + + **Sequential only.** One connection is open at a time. There are + no threads, no async, and no concurrency anywhere in this class: + ``accept`` on a cap that already holds an open connection returns + an ``Err`` telling the caller to ``close`` first. Capa's async + story is deliberately gated (``docs/design/async-feasibility.md``) + and ``Serve`` does not pre-empt it. A program serves one client, + finishes with it, and loops. + + **Nothing blocks forever.** ``accept`` and ``read`` are bounded by + ``_SERVE_ACCEPT_TIMEOUT_SECS`` / ``_SERVE_CONN_TIMEOUT_SECS`` and + return ``Err`` on expiry rather than hanging. + + Attenuation domain: the pair (bind address, port). An instance + carries either ``None`` (unrestricted authority, the fresh + capability supplied by ``main``) or a frozen set of RULES, each + parsed from a spec string: + + - ``"127.0.0.1:8080"`` - that address, that port + - ``"127.0.0.1:8000-8100"`` - that address, that inclusive port + range + - ``"127.0.0.1:*"`` - that address, any port + - ``"*:8080"`` - any address, that port + - ``"*:*"`` - any address, any port + + A bind is permitted only if it satisfies EVERY rule in the set, + the same conjunctive rule ``Fs`` uses for path prefixes. That is + what makes ``restrict_to`` narrowing-only: adding a rule can only + ever remove (addr, port) pairs from the permitted set, never add + one, so ``restrict_to("*:*")`` on an already-narrowed cap does not + restore anything. A spec that does not parse yields a rule nothing + satisfies (fail closed). + + The address is matched by exact string equality (or ``"*"``); + ``"127.0.0.1"`` and ``"localhost"`` are different rules, and IPv6 + literals with their embedded colons are not expressible. Both are + documented limits rather than oversights: the alternative is + resolving names inside the attenuation check, which would make the + permitted set depend on DNS and therefore not be a static property + of the cap at all. + + Enforcement happens BEFORE the syscall: ``listen`` consults + ``allows`` and returns the deny ``Err`` without creating a socket, + so a denied address/port is never bound even momentarily. + + Socket state does NOT survive attenuation. ``restrict_to`` returns + a fresh, un-bound ``Serve``; a narrowed capability is something you + take before you listen, not a second handle onto an already-open + listener. + + **Python backend only.** ``wasi:sockets`` is not vendored and is + unreachable from the wasmtime bindings the Wasm hosts use, so a + program holding ``Serve`` is rejected with an explicit diagnostic + by BOTH Wasm-facing paths -- the emitter (``capa --wasm``) and WIT + generation (``capa --wit``, which never runs the emitter and so + needs its own check). They share one reachability scan, in + ``capa.ir._python_only_caps``. + + Methods: + + - ``restrict_to(spec: String) -> Serve``: attenuation + - ``allows(addr: String, port: Int) -> Bool``: query without IO + - ``listen(addr: String, port: Int) -> Result``: + bind + listen. Port ``0`` asks the OS for an ephemeral port + (permitted only if the restriction admits port 0). + - ``local_port() -> Result``: the port actually + bound, which is how a caller learns the ephemeral port + - ``accept() -> Result``: wait for one client and + return a connection id + - ``recv(conn: Int, max_bytes: Int) -> Result, IoError>``: + up to ``max_bytes`` bytes as ints in ``0..=255``. An EMPTY list + means the peer closed (EOF). + - ``send(conn: Int, bytes: List) -> Result``: + send every byte; each element is masked with ``& 0xFF``. On + ``Ok`` the whole payload went out. On ``Err`` the number of bytes + already transmitted is UNSPECIFIED (a timeout can strike after a + prefix has been pushed), so a retry may duplicate data; close the + connection instead. See :meth:`send`. + - ``close(conn: Int) -> Result``: close one + connection + - ``stop() -> Result``: close the listener and any + open connection + + Spelled ``recv`` / ``send`` rather than ``read`` / ``write``: they + are the standard socket verbs, and ``write`` specifically is + unavailable because ``Fs.write`` already owns that name in the IFC + sink table, whose per-capability attribution is sound only while + each sink method name belongs to exactly one capability (see + ``capa.analyzer._ifc_summary``). + + Information flow: bytes arriving from a client are ``@public``. + Capa's IFC lattice models CONFIDENTIALITY, not integrity or taint, + so ``@public`` here says "this is not a secret whose disclosure + the lattice must prevent" - it emphatically does NOT say the bytes + are trustworthy. Validate inbound data like you would anywhere + else. Labelling it ``@secret`` would make echoing a request back + to its own sender an IFC violation, which is the normal case for a + server, so the useful signal would drown in noise. + """ + + __slots__ = ("_rules", "_listener", "_conns", "_next_conn") + + def __init__(self, _rules=None): + # ``_rules`` is either None (unrestricted) or a frozenset of + # ``(host, port_lo, port_hi)`` triples produced by + # ``_parse_rule``. + self._rules = _rules + self._listener = None + self._conns: dict[int, Any] = {} + # Connection ids start at 1 so 0 is never a live connection, + # matching the "0 is the no-cap sentinel" convention of the + # Wasm cap handle table. + self._next_conn = 1 + + # ---- attenuation ------------------------------------------------ + + @staticmethod + def _parse_rule(spec: str): + """Parse an attenuation spec into ``(host, lo, hi)``. + + Returns ``_SERVE_DENY_ALL_RULE`` for anything that does not + parse, so a typo narrows to nothing instead of being dropped. + """ + if not isinstance(spec, str) or ":" not in spec: + return _SERVE_DENY_ALL_RULE + host, _, ports = spec.rpartition(":") + if not host: + return _SERVE_DENY_ALL_RULE + if ports == "*": + return (host, 0, 65535) + if "-" in ports: + lo_s, _, hi_s = ports.partition("-") + else: + lo_s = hi_s = ports + try: + lo, hi = int(lo_s), int(hi_s) + except ValueError: + return _SERVE_DENY_ALL_RULE + if not (0 <= lo <= 65535 and 0 <= hi <= 65535) or lo > hi: + return _SERVE_DENY_ALL_RULE + return (host, lo, hi) + + def restrict_to(self, spec: str) -> "Serve": + existing = self._rules or frozenset() + return Serve(_rules=existing | {self._parse_rule(spec)}) + + @staticmethod + def _is_valid_port(port) -> bool: + """A port that a bind could conceivably accept: an ``int`` (not + a ``bool`` -- ``True`` is an ``int`` in Python and would + otherwise read as port 1) in ``0..=65535``.""" + if isinstance(port, bool) or not isinstance(port, int): + return False + return 0 <= port <= 65535 + + def allows(self, addr: str, port: int) -> bool: + # Port validity is checked BEFORE the unrestricted early return, + # so ``allows`` means exactly what it documents: "would + # ``listen(addr, port)`` be permitted". Previously the checks + # sat after it, so an UNRESTRICTED cap answered True for port + # 65536 while ``listen`` refused -- fail-safe, but the query + # and the operation disagreed, which is precisely what an + # ``allows``-style predicate exists to prevent. + if not self._is_valid_port(port): + return False + if self._rules is None: + return True + for host, lo, hi in self._rules: + if host != "*" and host != addr: + return False + if not (lo <= port <= hi): + return False + return True + + def _rules_repr(self): + if self._rules is None: + return "unrestricted" + return sorted( + f"{host}:{lo}-{hi}" for host, lo, hi in self._rules + ) + + def _deny(self, addr: str, port: int): + return Err(IoError( + f"Serve capability does not permit listening on " + f"{addr!r} port {port}", + f"current restrictions: {self._rules_repr()}", + )) + + # ---- listener --------------------------------------------------- + + def listen(self, addr: str, port: int) -> "Result[None, IoError]": + # An out-of-range port is reported as what it is, ahead of the + # attenuation check, so the diagnostic does not depend on + # whether the cap happens to be restricted. Without this, an + # unrestricted cap said "bind(): port must be 0-65535" while a + # restricted one said "does not permit ... current + # restrictions: unrestricted" -- two different messages for one + # invalid input, the second self-contradictory. + if not self._is_valid_port(port): + return Err(IoError( + f"invalid port {port!r}", + "a port must be an integer in 0-65535 (0 asks the OS " + "for an ephemeral port)", + )) + # Attenuation is checked BEFORE any socket exists, so a denied + # address/port is never bound, not even transiently. + if not self.allows(addr, port): + return self._deny(addr, port) + if self._listener is not None: + return Err(IoError( + "Serve is already listening", + "call stop() before listening on another address", + )) + sock = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + # No SO_REUSEADDR: on Windows it permits a second process + # to steal a bound port, which would undermine the whole + # point of a capability that names an address and port. + sock.settimeout(_SERVE_ACCEPT_TIMEOUT_SECS) + sock.bind((addr, port)) + sock.listen(_SERVE_BACKLOG) + except (OSError, ValueError, OverflowError) as e: + if sock is not None: + sock.close() + return Err(IoError( + f"failed to listen on {addr!r} port {port}", str(e), + )) + self._listener = sock + return Ok(None) + + def local_port(self) -> "Result[int, IoError]": + if self._listener is None: + return Err(IoError( + "Serve is not listening", + "call listen(addr, port) first", + )) + try: + return Ok(int(self._listener.getsockname()[1])) + except (OSError, IndexError, ValueError) as e: + return Err(IoError("failed to read the bound port", str(e))) + + def accept(self) -> "Result[int, IoError]": + if self._listener is None: + return Err(IoError( + "Serve is not listening", + "call listen(addr, port) first", + )) + if self._conns: + return Err(IoError( + "Serve handles one connection at a time", + "close the open connection before accepting another", + )) + try: + self._listener.settimeout(_SERVE_ACCEPT_TIMEOUT_SECS) + conn, _peer = self._listener.accept() + except TimeoutError: + # ``socket.timeout`` is an alias of ``TimeoutError`` and a + # subclass of OSError, so this arm must precede the OSError + # arm or a timeout would be reported as a generic failure. + return Err(IoError( + "accept timed out", + f"{_SERVE_ACCEPT_TIMEOUT_SECS:g}s elapsed with no " + f"inbound connection", + )) + except OSError as e: + return Err(IoError("accept failed", str(e))) + conn.settimeout(_SERVE_CONN_TIMEOUT_SECS) + handle = self._next_conn + self._next_conn += 1 + self._conns[handle] = conn + return Ok(handle) + + # ---- connection IO ---------------------------------------------- + + def _conn_or_none(self, conn: int): + if isinstance(conn, bool) or not isinstance(conn, int): + return None + return self._conns.get(conn) + + @staticmethod + def _unknown_conn(conn): + return Err(IoError( + "unknown connection", + f"no open connection with id {conn!r}", + )) + + def recv(self, conn: int, max_bytes: int): + sock = self._conn_or_none(conn) + if sock is None: + return self._unknown_conn(conn) + if isinstance(max_bytes, bool) or not isinstance(max_bytes, int): + return Err(IoError( + "read size must be an integer", f"got {max_bytes!r}", + )) + if max_bytes <= 0: + return Err(IoError( + "read size must be positive", f"got {max_bytes}", + )) + try: + data = sock.recv(min(max_bytes, _SERVE_MAX_READ)) + except TimeoutError: + return Err(IoError( + "read timed out", + f"{_SERVE_CONN_TIMEOUT_SECS:g}s elapsed with no data", + )) + except OSError as e: + return Err(IoError("read failed", str(e))) + # An empty list is EOF: the peer closed its side. Bytes are + # ints in 0..=255, the same shape every other Capa byte API + # uses. + return Ok(CapaList([b & 0xFF for b in data])) + + def send(self, conn: int, data) -> "Result[None, IoError]": + """Send every byte of ``data`` on ``conn``. + + ``Ok`` means the whole payload was handed to the kernel. + + .. warning:: + **On ``Err``, the number of bytes already transmitted is + UNSPECIFIED.** This delegates to ``socket.sendall``, which + can push a prefix of the payload and only then hit the + timeout or the error; it does not report how far it got, + and neither does this method. So an ``Err`` does NOT mean + "nothing was sent". + + A caller that simply retries after an ``Err`` can therefore + duplicate a prefix on the wire. There is no safe generic + recovery: treat a failed ``send`` as having put the + connection into an indeterminate state and ``close`` it, + rather than attempting to resume. If a protocol needs + resumable writes it must carry its own sequencing, which is + exactly the kind of thing that belongs in a Capa library + rather than in this capability. + """ + sock = self._conn_or_none(conn) + if sock is None: + return self._unknown_conn(conn) + try: + payload = bytes(int(b) & 0xFF for b in data) + except (TypeError, ValueError) as e: + return Err(IoError( + "send payload is not a list of byte values", str(e), + )) + try: + sock.sendall(payload) + except TimeoutError: + return Err(IoError( + "send timed out", + f"{_SERVE_CONN_TIMEOUT_SECS:g}s elapsed; the number of " + f"bytes already transmitted is unspecified", + )) + except OSError as e: + return Err(IoError( + "send failed", + f"{e}; the number of bytes already transmitted is " + f"unspecified", + )) + return Ok(None) + + def close(self, conn: int) -> "Result[None, IoError]": + sock = self._conn_or_none(conn) + if sock is None: + return self._unknown_conn(conn) + del self._conns[conn] + _close_quietly(sock) + return Ok(None) + + def stop(self) -> "Result[None, IoError]": + if self._listener is None: + return Err(IoError( + "Serve is not listening", + "call listen(addr, port) first", + )) + for sock in self._conns.values(): + _close_quietly(sock) + self._conns.clear() + _close_quietly(self._listener) + self._listener = None + return Ok(None) + + +def _close_quietly(sock) -> None: + """Close a socket, ignoring the errors a close can raise on an + already-broken peer. Teardown must not manufacture a new failure + for the caller: ``close`` / ``stop`` report success once the + program can no longer use the socket, which is true either way.""" + try: + sock.shutdown(socket.SHUT_RDWR) + except OSError: + pass + try: + sock.close() + except OSError: + pass + + class Proc: """Capability for sandboxed subprocess execution, with first- class attenuation that mirrors :class:`Net` (the underlying diff --git a/capa/runtime/_trace.py b/capa/runtime/_trace.py index d27ba0a..bad89e5 100644 --- a/capa/runtime/_trace.py +++ b/capa/runtime/_trace.py @@ -59,7 +59,36 @@ def records() -> list[tuple[str, str, Optional[str]]]: ``(class, method, first_arg)`` triples, oldest first. ``first_arg`` is the first positional argument of the call when it is a string and ``None`` otherwise. This is the form - the attenuation-honoured invariant consumes.""" + the attenuation-honoured invariant consumes. + + .. note:: + **The record is one string wide, so a capability whose + restriction domain is not a single string is UNDER-RECORDED.** + ``Serve`` (2026-07) is the first such capability: its domain is + the pair (bind address, port), and ``listen("127.0.0.1", 8080)`` + records only ``"127.0.0.1"`` -- the port is lost. A replay over + these records therefore cannot decide whether a ``Serve`` + restriction was honoured, and must not pretend to. + + Deliberately NOT widened. The triple shape is consumed by the + property suite's ``_attenuation_violations`` / + ``_ATTENUATORS`` / ``_GATED_OPS`` machinery, and changing it + would churn a soundness harness that currently covers seven + capabilities correctly, to buy coverage for one that is tested + directly and more thoroughly by ``TestServeAttenuation`` / + ``TestServeAttenuationIsEnforced`` in + tests/test_serve_capability.py (including the check that a + denied bind leaves the port genuinely unbound, which a trace + replay could not establish at all). + + The consequence to keep in mind: ``Serve`` appears in + :func:`classes_used` (so the cap-subset invariant does cover + it), but it is absent from ``_ATTENUATORS`` / ``_GATED_OPS``, + so the attenuation-honoured invariant silently skips it. That + is a known gap, not an oversight. A future capability with a + multi-part restriction domain should widen this record rather + than inherit the gap. + """ return list(_records) @@ -84,9 +113,20 @@ def enable() -> None: from . import _capabilities as caps + # ``Unsafe`` is absent because it is method-less (it exists only as + # proof of authority the checker verifies), so there is nothing to + # wrap. + # + # ``Serve`` IS wrapped, so ``classes_used()`` sees it, but note it + # is NOT covered by the property suite's attenuation-honoured + # invariant: that replay models a restriction as a single string + # argument (``_ATTENUATORS`` / ``_GATED_OPS`` in + # tests/test_properties.py), and ``Serve.allows`` takes an address + # AND a port. Its attenuation is covered directly instead, by + # TestServeAttenuation* in tests/test_serve_capability.py. for cls in ( caps.Stdio, caps.Fs, caps.Env, caps.Clock, caps.Random, caps.Net, - caps.Db, caps.Proc, + caps.Db, caps.Proc, caps.Serve, ): _wrap_class(cls) diff --git a/capa/transpiler/__init__.py b/capa/transpiler/__init__.py index 970b8ec..dbab0a2 100644 --- a/capa/transpiler/__init__.py +++ b/capa/transpiler/__init__.py @@ -77,7 +77,7 @@ from dataclasses import dataclass, field from capa.runtime import ( Ok, Err, Result, Some, None_, Option, IoError, - Stdio, Fs, Env, Clock, Random, Net, Proc, Db, Unsafe, + Stdio, Fs, Env, Clock, Random, Net, Proc, Db, Serve, Unsafe, CapaList, CapaRange, CapaSet, _NoneType, parse_int, parse_float, to_float, to_int, diff --git a/capa/typesys.py b/capa/typesys.py index c9e8f4f..f16bbf8 100644 --- a/capa/typesys.py +++ b/capa/typesys.py @@ -140,7 +140,7 @@ class _TyUnknownSingleton(Ty): # so parameters of type Stdio, Fs, etc. are accepted as annotations. CAPABILITY_NAMES: frozenset[str] = frozenset({ "Stdio", "Fs", "Net", "Env", "Proc", "Clock", "Random", "Db", - "Unsafe", + "Serve", "Unsafe", }) diff --git a/docs/getting-started.md b/docs/getting-started.md index 14cd0b8..19d0a86 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -103,7 +103,7 @@ For a progressive 10-chapter introduction to the language, open | `capa --check file.capa` | Type-check only (do not run). | | `capa --transpile file.capa` | Print the generated Python code. | | `capa --ir [--run|--transpile] file.capa` | Use the CIR (capability-aware IR) pipeline instead of the direct legacy transpiler. Same observable output; falls back to legacy when CIR lowering hits an unsupported construct. | -| `capa --wasm --run file.capa` | Compile via CIR -> WAT -> binary `.wasm` and execute on `wasmtime` with a Python host bridge. Requires `wasm-tools` and `wasmtime` on PATH. The Wasm backend has full generics and trait parity with the Python reference; the main remaining limits are a few stdlib surface gaps and a loud error if a trait is used as a `Map` key / `Set` element. See `examples/wasm/`. | +| `capa --wasm --run file.capa` | Compile via CIR -> WAT -> binary `.wasm` and execute on `wasmtime` with a Python host bridge. Requires `wasm-tools` and `wasmtime` on PATH. The Wasm backend has full generics and trait parity with the Python reference; the main remaining limits are a few stdlib surface gaps, a loud error if a trait is used as a `Map` key / `Set` element, and a loud rejection of programs reaching the Python-only `Serve` / `Unsafe` capabilities. See `examples/wasm/`. | | `capa --wasm --transpile file.capa` | Print the WAT (WebAssembly text format) the Wasm backend would compile. | | `capa --wasm -o X.wasm file.capa` | Save the assembled core module to `X.wasm`. Add `--component` to wrap it in a Component Model component (via `wasm-tools component new`). | | `capa --wit file.capa` | Emit the WIT spec describing the program's capability imports. Useful for inspecting what a Wasm component would request from its host. | @@ -130,7 +130,26 @@ fun main(stdio: Stdio) The `stdio` parameter is a *capability*, only functions that receive it can perform I/O. Other available capabilities: `fs` (filesystem), `env` (environment variables), `clock` (time), `random` (random -numbers). +numbers), `net` (outbound HTTP), `db` (SQLite), `proc` (subprocesses), +`serve` (inbound TCP), and `unsafe` (the Python escape hatch). Full +reference in [`stdlib.md`](stdlib.md). + +> **Two capabilities are Python-backend only: `Serve` and `Unsafe`.** +> This is a permanent stance, not a backlog item. `Serve` would need +> `wasi:sockets`, which is neither vendored nor reachable from the +> `wasmtime-py` bindings the Wasm hosts use; `Unsafe` grants raw +> pointer / FFI / memory-map primitives with no sandboxed Wasm +> equivalent. `capa --run` and `capa --check` handle both. `capa +> --wasm` refuses such a program at discovery time, before any +> lowering, and names every offending site: +> +> ``` +> capa: --wasm: the Serve capability is intentionally not supported on the Wasm backend (binding a listening socket needs wasi:sockets, which is neither vendored in capa/wasi_wit nor reachable from the wasmtime-py bindings the Wasm hosts are built on, so a guest can never be handed an inbound connection). Use the Python backend for these functions, or refactor to remove the Serve parameter. +> - main(serve: Serve) +> ``` +> +> Wasm serving is a known future item contingent on `wasi:sockets`, +> not a silent gap. ## 6. A recommended first "real" program diff --git a/docs/reference.md b/docs/reference.md index 90c515f..a217506 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -435,9 +435,11 @@ the scrutinee's type arguments. ### 6.1. What they are Capabilities are primitive types representing access to system -resources (`Stdio`, `Fs`, `Env`, `Clock`, `Random`, `Unsafe`). They -are only accessible via function parameters, there are no global -instances. +resources (`Stdio`, `Fs`, `Env`, `Clock`, `Random`, `Net`, `Db`, +`Proc`, `Serve`, `Unsafe`). They are only accessible via function +parameters, there are no global instances. `Serve` and `Unsafe` run +on the Python backend only; `capa --wasm` rejects a program whose +signatures reach either (see [`stdlib.md`](stdlib.md)). ### 6.2. The capability discipline (three layers) @@ -498,13 +500,29 @@ variable inherits the iterable's label; and a secret pushed / added / set into a mutable `List` / `Set` / `Map` taints the container. **Sources and sinks**. `env.get(...)` is secret-by-default (its -result is `@secret` with no annotation). The public sinks are -`Stdio.print` / `println` / `eprintln`, `Net.get` / `post`, -`Fs.write`, and `Db.exec` / `query`. A `@secret` value reaching a -sink-position argument is an information-flow violation: a warning by -default, a hard error inside a function annotated `@strict_ifc()` -(which also turns on implicit-flow checking, where a sink inside a -branch guarded by a secret condition is reported). +result is `@secret` with no annotation), and it is the *only* secret +source: no other built-in method produces `@secret` data without an +annotation. The public sinks are `Stdio.print` / `println` / +`eprintln`, `Net.get` / `post`, `Fs.write`, `Db.exec` / `query`, and +`Serve.send` (the payload argument only, not the connection id, which +the runtime issued rather than the program). A `@secret` value +reaching a sink-position argument is an information-flow violation: a +warning by default, a hard error inside a function annotated +`@strict_ifc()` (which also turns on implicit-flow checking, where a +sink inside a branch guarded by a secret condition is reported). + +`Serve.recv(...)` is the one inbound *source*, and it is `@public`. +**This lattice models confidentiality, not integrity or taint.** +`@public` on an inbound request asserts only "this is not a secret +whose disclosure the analysis must prevent". It asserts **nothing** +about the data being trustworthy, well-formed, or safe to act on, and +an inbound request is attacker-controlled: validate it as you would +in any language. Labelling it `@secret` would encode an integrity +property in a confidentiality lattice, whose immediate effect is that +echoing a request back to the client that sent it, the normal case +for a server, would be reported as a violation. Integrity / taint +tracking would be a second lattice, not a relabelling of this one. +See [`stdlib.md`](stdlib.md) under `Serve`. **Declassification**. `declassify(value, reason: "...")` is the single auditable secret-to-public bridge. It is identity at runtime diff --git a/docs/stdlib.md b/docs/stdlib.md index cd78e9f..5d51c77 100644 --- a/docs/stdlib.md +++ b/docs/stdlib.md @@ -665,12 +665,191 @@ fun main(proc: Proc, stdio: Stdio) Err(e) -> stdio.eprintln("${e}") ``` +### `Serve` + +Authority to listen on a TCP address and do I/O on inbound +connections. Every other capability reaches out; `Serve` is the one +that is reached. It is deliberately *connection-level*, not +HTTP-level: the trusted runtime binds, accepts, and moves bytes, so +protocol parsing is ordinary Capa code in a library rather than +trusted-runtime code. + +> **Python backend only.** `Serve` is not available on the Wasm +> backends, and this is a permanent stance rather than a backlog +> item: binding a listening socket needs `wasi:sockets`, which is +> neither vendored in `capa/wasi_wit/deps/` nor reachable from the +> `wasmtime-py` bindings the Wasm hosts are built on. Both +> Wasm-facing paths refuse a program whose signatures reach `Serve`: +> `capa --wasm` at discovery time, before any lowering, and +> `capa --wit`, which never runs the emitter and so carries its own +> check. Each lists **every** offending site, so one compile shows the +> full extent of the refactor: +> +> ``` +> capa: --wasm: the Serve capability is intentionally not supported on the Wasm backend (binding a listening socket needs wasi:sockets, which is neither vendored in capa/wasi_wit nor reachable from the wasmtime-py bindings the Wasm hosts are built on, so a guest can never be handed an inbound connection). Use the Python backend for these functions, or refactor to remove the Serve parameter. +> - main(serve: Serve) +> ``` +> +> Use `capa --run` / `capa --check` (the Python backend) for these +> functions, or move the `Serve`-holding code into a separate program. +> This is the same treatment `Unsafe` gets, driven by the same +> `PYTHON_ONLY_CAPS` registry in +> [`capa/ir/_capa_types.py`](../capa/ir/_capa_types.py). Wasm serving +> is a known future item contingent on `wasi:sockets`, not a silent +> gap. + +| Method | Type | Description | +|---|---|---| +| `restrict_to(spec: String)` | `Serve` | Attenuate: a fresh, **un-bound** `Serve` carrying the current rule set **plus** the rule parsed from `spec`. Monotonic, restrictions only narrow. | +| `allows(addr: String, port: Int)` | `Bool` | Would `listen(addr, port)` be permitted? Performs no I/O. Takes the same two arguments as `listen` (unlike the one-`String` `allows` of `Fs` / `Net` / `Db` / `Proc`) precisely so the question and the answer line up. | +| `listen(addr: String, port: Int)` | `Result` | Bind and start listening. Returns `Err` *before* any socket exists when the restriction denies `(addr, port)`. Port `0` asks the OS for an ephemeral port. | +| `local_port()` | `Result` | The port actually bound, which is how a caller learns the port chosen for `listen(addr, 0)`. | +| `accept()` | `Result` | Wait for one client and return a **connection id** (ids start at 1). Bounded by a 30s timeout. | +| `recv(conn: Int, max_bytes: Int)` | `Result, IoError>` | Read up to `max_bytes` bytes, each an `Int` in `0..=255`. An **empty list is EOF** (the peer closed). Bounded by a 30s timeout. The runtime additionally caps a single read at 1 MiB, so a larger `max_bytes` does not read more. | +| `send(conn: Int, bytes: List)` | `Result` | Send every byte; each element is masked with `& 0xFF`, so `[0x41, 0x1FF, -1]` goes out as `41 ff ff`. Same byte shape `String.bytes()` produces. On `Ok` the whole payload was handed to the kernel. **On `Err`, how many bytes already went out is unspecified** (see below). | +| `close(conn: Int)` | `Result` | Close one connection. Frees the slot, so the next `accept` can proceed. | +| `stop()` | `Result` | Close the listener and any open connection. | + +Every fallible method returns `Result`; nothing panics +and nothing blocks forever. A port outside `0..=65535` is refused +rather than passed to the OS, and `allows` answers `False` for it. + +**A failed `send` is not "nothing was sent".** `send` delegates to +`socket.sendall`, which can push a prefix of the payload and only +then hit the timeout or the error, and it does not report how far it +got. So an `Err` from `send` leaves the number of transmitted bytes +**unspecified**, and a caller that simply retries can duplicate a +prefix on the wire. There is no safe generic recovery: treat a failed +`send` as having put the connection into an indeterminate state and +`close` it. A protocol that needs resumable writes must carry its own +sequencing, which is exactly the kind of thing that belongs in a Capa +library rather than in the capability. + +**Sequential only, and enforced.** One connection is open at a time. +There are no threads, no async, and no concurrency anywhere in the +capability. Calling `accept` while a connection is still open does +not queue and does not fork, it returns +`Err(IoError("Serve handles one connection at a time", "close the +open connection before accepting another"))`. `accept` and `recv` +are each bounded at 30 seconds and return `Err("accept timed out")` / +`Err("read timed out")` rather than hanging. A `Serve` program +therefore serves one client, finishes with it, `close`s, and loops. +This is a considered position, not an unfinished corner: see +[`docs/design/async-feasibility.md`](design/async-feasibility.md), +whose recommendation is to defer concurrency until there is a +concrete I/O-bound driver **and** real GC has landed **and** there is +appetite to reopen the machine-checked noninterference proof, which +concurrency would reopen. + +**`@public` is a confidentiality statement, not a safety one.** Bytes +returned by `recv` are `@public`. Capa's information-flow lattice +models **confidentiality** (who may learn a value) and **not** +integrity or taint. `@public` on an inbound request asserts only +"this is not a secret whose disclosure the analysis must prevent". It +asserts **nothing** about the data being trustworthy, well-formed, or +safe to act on, and an inbound request is attacker-controlled. +Validate it exactly as you would in any other language. The reason +inbound data is not `@secret` is that it would put an integrity +property in a confidentiality lattice: echoing a request back to the +client that sent it, the most ordinary thing a server does, would be +reported as a violation, and the useful signal would drown in the +noise. `Serve.send` *is* a public sink (the payload argument only), +so a `@secret` value reaching it is reported. See +[`reference.md`](reference.md) section 6.4. + +**Attenuation.** The domain is the pair (bind address, port). A spec +string is one of: + +| Spec | Meaning | +|---|---| +| `"127.0.0.1:8080"` | that address, that port | +| `"127.0.0.1:8000-8100"` | that address, that inclusive port range | +| `"127.0.0.1:*"` | that address, any port | +| `"*:8080"` | any address, that port | +| `"*:*"` | any address, any port | + +A `Serve` received from `main` is unrestricted. `restrict_to` +*accumulates* a rule, and a bind is permitted only if it satisfies +**every** accumulated rule, the same conjunctive model `Fs` uses for +path prefixes. That is what makes narrowing monotonic: adding a rule +can only remove `(addr, port)` pairs from the permitted set, never +add one. In particular **`restrict_to("*:*")` on an +already-narrowed capability restores nothing**, and two disjoint +rules yield a capability that permits nothing. + +A spec that does not parse yields a rule nothing can satisfy: it +denies everything. That is deliberate. `restrict_to` returns a +`Serve`, not a `Result`, so there is nowhere to report a typo, and +silently ignoring one would **widen** authority. + +Enforcement runs *before* the syscall: `listen` consults `allows` and +returns the denial without ever creating a socket, so a denied +address or port is never bound, not even transiently +(`test_restricted_cap_cannot_bind_another_port` in +[`tests/test_serve_capability.py`](../tests/test_serve_capability.py) +asserts the refused port is still free afterwards). The denial +carries the current rule set: + +``` +Serve capability does not permit listening on '127.0.0.1' port 8080: current restrictions: ['127.0.0.1:9000-9010'] +``` + +Three honest limits: + +- **Addresses are matched by exact string equality** (or `"*"`). + `"127.0.0.1"` and `"localhost"` are different rules, and a rule for + one does not admit the other. The alternative, resolving names + inside the attenuation check, was rejected because it would make + the permitted set depend on DNS instead of being a static property + of the capability. +- **IPv4 only.** The runtime creates `AF_INET` sockets, so an IPv6 + address is not usable even though a spec mentioning one parses: + `listen("::1", 0)` fails with + `failed to listen on '::1' port 0` plus the OS resolver error. +- **Port `0` means "whatever the OS picks".** The check runs against + the port you *request*, so a restriction that admits port `0` + admits binding an arbitrary ephemeral port, which the same + restriction would refuse if you named it. Read the actual port back + with `local_port()`. + +Socket state does not survive attenuation: `restrict_to` returns a +fresh, un-bound `Serve`. A narrowed capability is something you take +*before* you listen, not a second handle onto an open listener. + +```capa +fun echo_once(serve: Serve, conn: Int) -> Result + let request = serve.recv(conn, 1024)? + return serve.send(conn, request) + +fun main(stdio: Stdio, serve: Serve) + let local = serve.restrict_to("127.0.0.1:8080") + match local.listen("127.0.0.1", 8080) + Ok(_) -> match local.accept() + Ok(conn) -> + match echo_once(local, conn) + Ok(_) -> stdio.println("echoed") + Err(e) -> stdio.eprintln("${e}") + let _ = local.close(conn) + let _ = local.stop() + Err(e) -> stdio.eprintln("${e}") + Err(e) -> stdio.eprintln("${e}") +``` + +See [`examples/wasm/serve_demo.capa`](../examples/wasm/serve_demo.capa) +for the ephemeral-port variant. + ### `Unsafe` Marker capability for crossing the Python boundary. Has no methods - its only role is to gate `py_import` and `py_invoke` (see "Python interoperability" above). +Like `Serve`, `Unsafe` is Python-backend only: `capa --wasm` rejects +a program whose signatures reach it, with the same shape of +diagnostic ("the Unsafe capability is intentionally not supported on +the Wasm backend (it grants raw pointer / FFI / memory-map primitives +that have no sandboxed Wasm equivalent) ..."). + ### User-defined capabilities Libraries can declare their own capabilities with the `capability` diff --git a/docs/tutorial.md b/docs/tutorial.md index 25b3af9..d70548f 100644 --- a/docs/tutorial.md +++ b/docs/tutorial.md @@ -359,7 +359,11 @@ Available capabilities: | `Env` | environment variables, args | | `Clock` | time, sleep | | `Random` | random numbers | -| `Unsafe` | crossing the Python boundary | +| `Net` | outbound HTTP | +| `Db` | SQLite databases | +| `Proc` | sandboxed subprocesses | +| `Serve` | inbound TCP: listen, accept, recv, send (Python backend only) | +| `Unsafe` | crossing the Python boundary (Python backend only) | ### Why capabilities? diff --git a/examples/wasm/serve_demo.capa b/examples/wasm/serve_demo.capa new file mode 100644 index 0000000..265dbbd --- /dev/null +++ b/examples/wasm/serve_demo.capa @@ -0,0 +1,42 @@ +// serve_demo.capa - the Serve capability: bind, accept one client, +// read its bytes, write a response, close. +// +// Serve is connection-level, not HTTP-level: the runtime moves bytes +// and nothing else, so any protocol handling is ordinary Capa code +// rather than trusted-runtime code. Bytes are List with each +// element in 0..=255, the same shape String.bytes() already returns. +// +// Serve is sequential: one connection at a time, no threads and no +// async. Both accept() and read() are bounded by a timeout and return +// Err rather than hanging forever. +// +// PYTHON BACKEND ONLY. The Wasm emitter rejects any signature that +// reaches Serve (wasi:sockets is not vendored and is not reachable +// from the wasmtime bindings the Wasm hosts use), so this file is +// listed in _EXCLUDED in tests/test_ir_wasm_parity.py. + +fun main(stdio: Stdio, serve: Serve) + // Attenuate FIRST, then listen. The narrowed cap admits only the + // loopback interface, and only an ephemeral port. + let local = serve.restrict_to("127.0.0.1:0") + + match local.listen("127.0.0.1", 0) + Ok(_) -> stdio.println("listening") + Err(e) -> stdio.println("listen failed: " + e.message) + + match local.accept() + Ok(conn) -> + match local.recv(conn, 1024) + Ok(request) -> stdio.println("read ${request.length()} bytes") + Err(e) -> stdio.println("read failed: " + e.message) + match local.send(conn, "pong".bytes()) + Ok(_) -> stdio.println("wrote pong") + Err(e) -> stdio.println("write failed: " + e.message) + match local.close(conn) + Ok(_) -> stdio.println("closed") + Err(e) -> stdio.println("close failed: " + e.message) + Err(e) -> stdio.println("accept failed: " + e.message) + + match local.stop() + Ok(_) -> stdio.println("stopped") + Err(e) -> stdio.println("stop failed: " + e.message) diff --git a/tests/test_cap_handles.py b/tests/test_cap_handles.py index e236b83..21f7568 100644 --- a/tests/test_cap_handles.py +++ b/tests/test_cap_handles.py @@ -14,6 +14,7 @@ BUILTIN_CAPS, ERASED_CAPS, HANDLE_BEARING_CAPS, + PYTHON_ONLY_CAPS, ) from capa.ir._emit_wit import _KNOWN_CAPABILITIES from capa.lsp.completion import _BUILTIN_CAPABILITIES @@ -278,15 +279,47 @@ def test_lsp_offers_every_builtin_capability(self): # the editor never suggested them. self.assertEqual(set(_BUILTIN_CAPABILITIES), set(CAPABILITY_NAMES)) - def test_wit_generator_knows_every_capability_except_unsafe(self): - # ``Unsafe`` is deliberately absent from the WIT generator's - # known set: it is the Python-only FFI escape hatch and the - # Wasm emitter rejects it up front, so it can never reach WIT + def test_wit_generator_knows_every_capability_except_python_only(self): + # The ``PYTHON_ONLY_CAPS`` are deliberately absent from the WIT + # generator's known set: the Wasm emitter rejects any signature + # reaching them up front, so they can never get as far as WIT # generation. Every OTHER built-in must be known, or its # interface would be silently skipped and the host would be # asked for an import it was never told to provide. + # + # Derived from ``PYTHON_ONLY_CAPS`` rather than a hard-coded + # ``{"Unsafe"}``: when ``Serve`` landed (2026-07) the literal + # spelling made this guard fire for the right reason but with + # the wrong remedy on offer ("teach WIT about Serve", when the + # correct answer is "Serve never reaches WIT"). Deriving keeps + # the exemption and its justification the same fact. + self.assertEqual( + set(_KNOWN_CAPABILITIES), set(BUILTIN_CAPS) - PYTHON_ONLY_CAPS, + ) + + def test_python_only_caps_are_a_subset_of_the_builtins(self): + # A stale name here would silently exempt nothing while + # reading as though it exempted something. + stray = PYTHON_ONLY_CAPS - BUILTIN_CAPS + self.assertEqual( + stray, set(), + "PYTHON_ONLY_CAPS names capabilities that are not " + f"built-ins: {sorted(stray)}", + ) + + def test_python_only_caps_are_erased_on_the_wasm_side(self): + # A cap the Wasm backend REJECTS must not also be declared + # handle-bearing: that would claim a Wasm lowering for codegen + # that never runs, and if the rejection ever developed a hole + # the cap would occupy an i32 slot with no root-handle entry + # and silently degrade to the Fs root. Erased fails loud + # instead. + misclassified = PYTHON_ONLY_CAPS & HANDLE_BEARING_CAPS self.assertEqual( - set(_KNOWN_CAPABILITIES), set(BUILTIN_CAPS) - {"Unsafe"}, + misclassified, set(), + "capabilities the Wasm backend rejects must be in " + f"ERASED_CAPS, not HANDLE_BEARING_CAPS: " + f"{sorted(misclassified)}", ) diff --git a/tests/test_ir_wasm.py b/tests/test_ir_wasm.py index eb8ca0a..be9421f 100644 --- a/tests/test_ir_wasm.py +++ b/tests/test_ir_wasm.py @@ -9330,6 +9330,29 @@ def _recipe(ty, method, caps, setup, expr): } +def _method_is_exempt_from_the_sweep(ty: str) -> bool: + """True for methods that can never reach the Wasm emitter, so a + discarded-call recipe for them could not be built (and would not + mean anything if it were). + + The sweep proves a DISCARDED call still balances the Wasm operand + stack. A capability the Wasm backend rejects outright at discovery + time has no emit path at all: any program exercising it raises + ``WasmEmissionError`` before a single instruction is produced, so + there is no stack to balance. + + Derived from ``PYTHON_ONLY_CAPS`` rather than spelled out, for the + same reason the WIT guard is: the exemption and its justification + should be one fact, not two that can drift. ``Serve`` (2026-07) is + the first such capability that HAS methods -- ``Unsafe`` is + method-less, which is why this guard never had to express the idea + before. This mirrors the ``py_import`` / ``py_invoke`` exemption in + ``_FREE_FN_EXEMPT`` directly above. + """ + from capa.ir._capa_types import PYTHON_ONLY_CAPS + return ty in PYTHON_ONLY_CAPS + + class _EmitVariant(typing.NamedTuple): """One flag-selected emit path for a discarded call. @@ -9473,6 +9496,7 @@ def test_discard_recipes_cover_every_builtin(self): for ty, entries in METHODS.items() for entry in entries if (ty, entry[0]) not in _DISCARD_RECIPES + and not _method_is_exempt_from_the_sweep(ty) ] self.assertEqual( missing_methods, [], @@ -9483,6 +9507,20 @@ def test_discard_recipes_cover_every_builtin(self): "the Wasm operand stack." ), ) + # The exemption must not become a hiding place: a recipe for a + # Wasm-rejected capability would fail to emit, so assert none + # was written rather than silently tolerating one. + exempt_with_recipes = [ + key for key in _DISCARD_RECIPES + if _method_is_exempt_from_the_sweep(key[0]) + ] + self.assertEqual( + exempt_with_recipes, [], + msg=( + "a discarded-call recipe exists for a capability the " + "Wasm backend rejects outright; it can never emit" + ), + ) missing_free = [ name for name in FREE_FUNCTIONS if name not in _FREE_FN_RECIPES and name not in _FREE_FN_EXEMPT diff --git a/tests/test_ir_wasm_parity.py b/tests/test_ir_wasm_parity.py index 1856a2e..3b0e9ff 100644 --- a/tests/test_ir_wasm_parity.py +++ b/tests/test_ir_wasm_parity.py @@ -841,6 +841,18 @@ # here so a future contributor doesn't accidentally widen the # parity list without thinking about the divergence. _EXCLUDED: dict[str, str] = { + "serve_demo.capa": ( + "The Serve capability is Python-backend-only, so there is no " + "Wasm side to be at parity WITH: binding a listening socket " + "needs wasi:sockets, which is neither vendored in " + "capa/wasi_wit nor reachable from the wasmtime-py bindings " + "the hosts are built on. The Wasm emitter rejects any " + "signature reaching Serve up front (same treatment as " + "Unsafe); that rejection is covered by " + "TestWasmRejectsServe in tests/test_serve_capability.py, and " + "the program's actual behaviour by TestServeEndToEnd in the " + "same file, which drives it with a real socket client." + ), "clock_demo.capa": ( "Clock.now_secs / now_monotonic are time-dependent; their " "values differ between back-to-back runs even on one backend." diff --git a/tests/test_serve_capability.py b/tests/test_serve_capability.py new file mode 100644 index 0000000..c007c95 --- /dev/null +++ b/tests/test_serve_capability.py @@ -0,0 +1,1071 @@ +"""Tests for the ``Serve`` capability (2026-07): the authority to +listen on a network address and accept inbound connections. + +``Serve`` is the tenth built-in capability and the language's first +INBOUND authority. It is connection-level (listen / accept / recv / +send / close), Python-backend-only, and sequential: one open +connection at a time, no threads and no async inside the runtime. + +Coverage, in the order the acceptance criteria were set: + +- END TO END: a real Capa program binds an ephemeral loopback port, + accepts one connection, reads the request bytes and writes a + response, driven by a Python ``socket`` client that asserts the + exact bytes on the wire. +- ATTENUATION: narrowing is intersection-only (adding a rule can never + widen), and the denial is REAL -- a cap restricted to one port + cannot bind another, and the port it was refused stays unbound, so + the deny is not a relabelling of a bind that actually happened. +- SEQUENTIAL / BOUNDED: accept before listen, a second accept while a + connection is open, and the accept timeout all return ``Err``. +- WASM REJECTION: the emitter names every offending site. This test + deliberately does NOT need wasmtime (the rejection happens at emit + time), so it runs on the no-wasm CI job, which is the job most + likely to regress it. +- MANIFEST / SBOM / POLICY: ``Serve`` appears on exactly the functions + that hold it, reaches the composed SBOM, and a ``forbid-capability`` + policy over it fires. These layers derive from ``CAPABILITY_NAMES`` + and needed no code changes; these tests verify that rather than + assuming it. + +NO wasm tooling is imported at module level anywhere in this file, so +the module always COLLECTS under ``python -m unittest discover`` on a +runner without the ``wasm`` extra installed. +""" + +import io +import socket +import tempfile +import threading +import unittest +from contextlib import redirect_stdout +from pathlib import Path + +from capa import Lexer, Parser, analyze, transpile +from capa.builtins import METHODS +from capa.ir import lower +from capa.ir._capa_types import ( + BUILTIN_CAPS, + ERASED_CAPS, + HANDLE_BEARING_CAPS, + PYTHON_ONLY_CAPS, +) +from capa.loader import ModuleLoader +from capa.manifest import ( + build_composed_sbom, build_manifest, evaluate_policies, find_policy_file, + read_policy_file, +) +from capa.runtime._capabilities import Serve +from capa.typesys import CAPABILITY_NAMES + + +_EXAMPLES = Path(__file__).resolve().parent.parent / "examples" / "wasm" + + +def _parse_and_analyze(src: str, filename: str = "t.capa"): + """Front end only: lex, parse, analyze, and fail loudly on a + diagnostic. Same shape as the parity suite's helper.""" + tokens = Lexer(src).lex() + module = Parser(tokens, source=src).parse_module() + result = analyze(module, source=src, filename=filename) + if not result.ok: + raise AssertionError(f"analyzer errors: {result.errors}") + return module, result + + +def _free_port() -> int: + """Bind an ephemeral loopback port, note it, and release it. + + Used to pick a port to TEMPLATE INTO a Capa program (the program + itself cannot report its port back to the test without a channel + we would have to build). There is a theoretical race between + release and the program's bind, but on loopback with no other + listener it does not fire in practice, and a genuine collision + surfaces as a loud bind failure rather than a silent pass. + """ + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + s.bind(("127.0.0.1", 0)) + return int(s.getsockname()[1]) + finally: + s.close() + + +def _port_is_free(port: int) -> bool: + """True iff nothing is listening on ``port``. This is what turns + 'the deny returned Err' into 'the deny actually prevented a bind'.""" + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + s.settimeout(0.5) + return s.connect_ex(("127.0.0.1", port)) != 0 + finally: + s.close() + + +def _run_capa(source: str, filename: str, *cap_args) -> str: + """Transpile ``source`` and run its ``main`` with the given caps, + returning whatever ``main`` printed. Mirrors the in-process + transpile+exec harness the parity suite uses (fast, no + subprocess). + + ``__name__`` is deliberately NOT ``"__main__"`` here: the + transpiled bootstrap would then construct its own capabilities and + run ``main`` at import, and this caller needs to supply the caps + itself (and to know when ``main`` returned).""" + module, result = _parse_and_analyze(source, filename) + code = transpile( + module, filename=filename, + types=result.types, bindings=result.bindings, + ) + ns: dict = {"__name__": filename} + exec(compile(code, filename, "exec"), ns) + buf = io.StringIO() + with redirect_stdout(buf): + ns["main"](*cap_args) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# END TO END: a real Capa program serving a real client +# --------------------------------------------------------------------------- + + +_ECHO_SERVER = '''\ +fun main(stdio: Stdio, serve: Serve) + let local = serve.restrict_to("127.0.0.1:{port}") + match local.listen("127.0.0.1", {port}) + Ok(_) -> stdio.println("listening") + Err(e) -> stdio.println("listen failed: " + e.message) + match local.accept() + Ok(conn) -> + match local.recv(conn, 1024) + Ok(request) -> stdio.println("read ${{request.length()}} bytes") + Err(e) -> stdio.println("read failed: " + e.message) + match local.send(conn, "pong".bytes()) + Ok(_) -> stdio.println("wrote pong") + Err(e) -> stdio.println("write failed: " + e.message) + match local.close(conn) + Ok(_) -> stdio.println("closed") + Err(e) -> stdio.println("close failed: " + e.message) + Err(e) -> stdio.println("accept failed: " + e.message) + match local.stop() + Ok(_) -> stdio.println("stopped") + Err(e) -> stdio.println("stop failed: " + e.message) +''' + + +class TestServeEndToEnd(unittest.TestCase): + """The headline acceptance row: a Capa program is a real server + that a real (Python) client talks to over a real socket.""" + + def test_capa_program_serves_one_client(self): + from capa.runtime._capabilities import Stdio + + port = _free_port() + source = _ECHO_SERVER.format(port=port) + + out: dict = {} + + def run_server(): + try: + out["stdout"] = _run_capa( + source, "echo_server.capa", Stdio(), Serve(), + ) + except BaseException as e: # pragma: no cover + out["error"] = e + + t = threading.Thread(target=run_server, daemon=True) + t.start() + + # Give the program a moment to reach its bind, retrying rather + # than sleeping a fixed amount so the test is not timing-fragile. + client = None + deadline = 5.0 + step = 0.02 + waited = 0.0 + while waited < deadline: + c = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + c.settimeout(2.0) + if c.connect_ex(("127.0.0.1", port)) == 0: + client = c + break + c.close() + threading.Event().wait(step) + waited += step + self.assertIsNotNone( + client, f"Capa program never bound 127.0.0.1:{port}", + ) + + try: + client.sendall(b"ping") + response = client.recv(64) + finally: + client.close() + + t.join(timeout=10.0) + self.assertFalse(t.is_alive(), "the Capa program did not finish") + self.assertNotIn("error", out, f"server raised: {out.get('error')!r}") + + # The bytes the client actually received off the wire. + self.assertEqual(response, b"pong") + # And the program's own view of the exchange. + self.assertEqual( + out["stdout"].splitlines(), + [ + "listening", + "read 4 bytes", + "wrote pong", + "closed", + "stopped", + ], + ) + + +# --------------------------------------------------------------------------- +# ATTENUATION +# --------------------------------------------------------------------------- + + +class TestServeAttenuation(unittest.TestCase): + """``restrict_to`` narrows and only narrows, and the narrowing is + enforced before the syscall.""" + + def test_unrestricted_allows_anything(self): + s = Serve() + self.assertTrue(s.allows("127.0.0.1", 8080)) + self.assertTrue(s.allows("0.0.0.0", 1)) + + def test_exact_addr_port_rule(self): + s = Serve().restrict_to("127.0.0.1:8080") + self.assertTrue(s.allows("127.0.0.1", 8080)) + self.assertFalse(s.allows("127.0.0.1", 8081)) + self.assertFalse(s.allows("0.0.0.0", 8080)) + + def test_port_range_rule(self): + s = Serve().restrict_to("127.0.0.1:8000-8100") + self.assertTrue(s.allows("127.0.0.1", 8000)) + self.assertTrue(s.allows("127.0.0.1", 8050)) + self.assertTrue(s.allows("127.0.0.1", 8100)) + self.assertFalse(s.allows("127.0.0.1", 7999)) + self.assertFalse(s.allows("127.0.0.1", 8101)) + + def test_wildcards(self): + any_port = Serve().restrict_to("127.0.0.1:*") + self.assertTrue(any_port.allows("127.0.0.1", 1)) + self.assertTrue(any_port.allows("127.0.0.1", 65535)) + self.assertFalse(any_port.allows("10.0.0.1", 80)) + + any_addr = Serve().restrict_to("*:8080") + self.assertTrue(any_addr.allows("10.0.0.1", 8080)) + self.assertFalse(any_addr.allows("10.0.0.1", 8081)) + + def test_restrict_to_only_ever_narrows(self): + # Chaining intersects: the surviving authority is exactly the + # single port in BOTH rules. + s = Serve().restrict_to("127.0.0.1:8000-8100") + s = s.restrict_to("127.0.0.1:8080") + self.assertTrue(s.allows("127.0.0.1", 8080)) + self.assertFalse(s.allows("127.0.0.1", 8081)) + + # And a deliberate WIDENING attempt restores nothing: the + # earlier rules still have to be satisfied. + widened = s.restrict_to("*:*") + self.assertTrue(widened.allows("127.0.0.1", 8080)) + self.assertFalse(widened.allows("127.0.0.1", 8081)) + self.assertFalse(widened.allows("10.0.0.1", 8080)) + + def test_disjoint_rules_permit_nothing(self): + s = Serve().restrict_to("127.0.0.1:8080").restrict_to("127.0.0.1:9090") + self.assertFalse(s.allows("127.0.0.1", 8080)) + self.assertFalse(s.allows("127.0.0.1", 9090)) + + def test_malformed_spec_fails_closed(self): + # restrict_to returns a Serve, not a Result, so it cannot + # report a parse error. A spec that does not parse must DENY, + # because the alternative (ignore it) silently widens. + for spec in ("nonsense", "127.0.0.1:", "127.0.0.1:abc", + "127.0.0.1:99999", "127.0.0.1:100-50", ":8080", ""): + with self.subTest(spec=spec): + s = Serve().restrict_to(spec) + self.assertFalse(s.allows("127.0.0.1", 8080)) + self.assertFalse(s.allows("127.0.0.1", 0)) + + def test_out_of_range_and_non_int_ports_denied(self): + s = Serve().restrict_to("127.0.0.1:*") + self.assertFalse(s.allows("127.0.0.1", -1)) + self.assertFalse(s.allows("127.0.0.1", 65536)) + self.assertFalse(s.allows("127.0.0.1", True)) + + def test_allows_agrees_with_listen_on_invalid_ports(self): + # ``allows`` documents itself as "would listen(a, p) be + # permitted", so the two must not disagree. The port-validity + # check used to sit AFTER the unrestricted early return, so an + # unrestricted cap answered True for port 65536 while listen + # refused: fail-safe, but a query that contradicts the + # operation it describes defeats the purpose of having one. + for label, cap in ( + ("unrestricted", Serve()), + ("restricted", Serve().restrict_to("127.0.0.1:*")), + ): + for bad in (-1, 65536, True, "80"): + with self.subTest(cap=label, port=bad): + self.assertFalse(cap.allows("127.0.0.1", bad)) + r = cap.listen("127.0.0.1", bad) + self.assertTrue(r.is_err()) + # One consistent diagnostic regardless of whether + # the cap is restricted; in particular never the + # self-contradictory "does not permit ... current + # restrictions: unrestricted". + self.assertIn("invalid port", r.error.message) + + def test_valid_ports_still_allowed_on_an_unrestricted_cap(self): + # Guard against the validity check over-rejecting: 0 (ephemeral) + # and 65535 (the boundary) must both survive. + s = Serve() + for good in (0, 1, 8080, 65535): + with self.subTest(port=good): + self.assertTrue(s.allows("127.0.0.1", good)) + + def test_parent_is_not_mutated_by_restriction(self): + parent = Serve() + child = parent.restrict_to("127.0.0.1:8080") + self.assertTrue(parent.allows("10.0.0.1", 9999)) + self.assertFalse(child.allows("10.0.0.1", 9999)) + + +class TestServeAttenuationIsEnforced(unittest.TestCase): + """The denial is real: a refused bind never happens, rather than + happening and being relabelled.""" + + def test_restricted_cap_cannot_bind_another_port(self): + allowed = _free_port() + forbidden = _free_port() + self.assertNotEqual(allowed, forbidden) + + s = Serve().restrict_to(f"127.0.0.1:{allowed}") + + result = s.listen("127.0.0.1", forbidden) + self.assertTrue(result.is_err(), "the forbidden bind was permitted") + err = result.error + self.assertIn("does not permit", err.message) + self.assertIn(str(forbidden), err.message) + + # The load-bearing assertion: nothing is listening on the port + # we were refused. A deny that merely returned Err after + # binding would leave a live listener here. + self.assertTrue( + _port_is_free(forbidden), + "the denied port was actually bound; the deny was a " + "relabelling, not an enforcement", + ) + + # The permitted port still works, so the cap was narrowed and + # not simply broken. + ok = s.listen("127.0.0.1", allowed) + self.assertTrue(ok.is_ok(), f"allowed bind failed: {ok}") + try: + self.assertEqual(s.local_port().unwrap(), allowed) + finally: + s.stop() + + def test_restricted_cap_cannot_bind_another_address(self): + port = _free_port() + s = Serve().restrict_to(f"127.0.0.1:{port}") + result = s.listen("0.0.0.0", port) + self.assertTrue(result.is_err()) + self.assertIn("does not permit", result.error.message) + + def test_narrow_range_forbids_ephemeral_port_zero(self): + # Port 0 asks the OS to choose. A cap narrowed to a range that + # excludes 0 must refuse, because otherwise the OS's choice + # would escape the restriction. + s = Serve().restrict_to("127.0.0.1:8000-8100") + self.assertFalse(s.allows("127.0.0.1", 0)) + self.assertTrue(s.listen("127.0.0.1", 0).is_err()) + + def test_ephemeral_bind_allowed_when_restriction_admits_zero(self): + s = Serve().restrict_to("127.0.0.1:0") + result = s.listen("127.0.0.1", 0) + self.assertTrue(result.is_ok(), f"ephemeral bind failed: {result}") + try: + self.assertGreater(s.local_port().unwrap(), 0) + finally: + s.stop() + + +# --------------------------------------------------------------------------- +# SEQUENTIAL SEMANTICS AND BOUNDED WAITS +# --------------------------------------------------------------------------- + + +class TestServeLifecycle(unittest.TestCase): + + def test_accept_before_listen_is_an_error(self): + r = Serve().accept() + self.assertTrue(r.is_err()) + self.assertIn("not listening", r.error.message) + + def test_local_port_before_listen_is_an_error(self): + self.assertTrue(Serve().local_port().is_err()) + + def test_stop_before_listen_is_an_error(self): + self.assertTrue(Serve().stop().is_err()) + + def test_double_listen_is_an_error(self): + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + try: + r = s.listen("127.0.0.1", 0) + self.assertTrue(r.is_err()) + self.assertIn("already listening", r.error.message) + finally: + s.stop() + + def test_recv_and_send_on_unknown_connection_are_errors(self): + s = Serve() + for r in (s.recv(1, 16), s.send(1, [1, 2]), s.close(1)): + self.assertTrue(r.is_err()) + self.assertIn("unknown connection", r.error.message) + + def test_accept_timeout_returns_err_not_a_hang(self): + # Shrink the bound so the test is fast; the point is that the + # wait TERMINATES with an Err rather than blocking forever. + import capa.runtime._capabilities as caps + original = caps._SERVE_ACCEPT_TIMEOUT_SECS + caps._SERVE_ACCEPT_TIMEOUT_SECS = 0.1 + try: + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + try: + r = s.accept() + self.assertTrue(r.is_err()) + self.assertIn("timed out", r.error.message) + finally: + s.stop() + finally: + caps._SERVE_ACCEPT_TIMEOUT_SECS = original + + def test_one_connection_at_a_time(self): + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + port = s.local_port().unwrap() + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + client.settimeout(5.0) + client.connect(("127.0.0.1", port)) + conn = s.accept().unwrap() + + # A second accept while a connection is open is refused + # with an actionable message, NOT queued and NOT threaded. + second = s.accept() + self.assertTrue(second.is_err()) + self.assertIn( + "one connection at a time", second.error.message, + ) + + self.assertTrue(s.close(conn).is_ok()) + finally: + client.close() + s.stop() + + def test_recv_reports_eof_as_an_empty_list(self): + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + port = s.local_port().unwrap() + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + client.settimeout(5.0) + client.connect(("127.0.0.1", port)) + conn = s.accept().unwrap() + client.shutdown(socket.SHUT_WR) + self.assertEqual(list(s.recv(conn, 16).unwrap()), []) + finally: + client.close() + s.stop() + + def test_bytes_round_trip_as_ints_masked_to_a_byte(self): + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + port = s.local_port().unwrap() + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + client.settimeout(5.0) + client.connect(("127.0.0.1", port)) + conn = s.accept().unwrap() + + client.sendall(bytes([0, 1, 127, 128, 255])) + got = list(s.recv(conn, 16).unwrap()) + self.assertEqual(got, [0, 1, 127, 128, 255]) + for b in got: + self.assertTrue(0 <= b <= 255) + + # Out-of-range ints are masked with & 0xFF rather than + # rejected, matching the byte convention elsewhere. + self.assertTrue(s.send(conn, [256 + 65, 0x1FF]).is_ok()) + self.assertEqual(client.recv(8), bytes([65, 255])) + finally: + client.close() + s.stop() + + def test_send_error_discloses_that_transmitted_count_is_unspecified(self): + # ``sendall`` can push a PREFIX of the payload and only then + # raise, so an Err does not mean "nothing was sent". A caller + # that retries would duplicate the prefix. The error text has + # to say so, because the type (Result) cannot. + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + port = s.local_port().unwrap() + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + client.settimeout(5.0) + client.connect(("127.0.0.1", port)) + conn = s.accept().unwrap() + # Force the failure arm by closing the peer, then sending + # enough to overflow any kernel buffer that would otherwise + # absorb the write silently. + client.close() + r = s.send(conn, [65] * (4 << 20)) + if r.is_err(): + self.assertIn("unspecified", r.error.cause) + finally: + s.stop() + + def test_send_rejects_a_non_byte_payload(self): + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + port = s.local_port().unwrap() + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + client.settimeout(5.0) + client.connect(("127.0.0.1", port)) + conn = s.accept().unwrap() + r = s.send(conn, ["not a byte"]) + self.assertTrue(r.is_err()) + self.assertIn("not a list of byte values", r.error.message) + finally: + client.close() + s.stop() + + def test_recv_size_must_be_positive(self): + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + port = s.local_port().unwrap() + client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + client.settimeout(5.0) + client.connect(("127.0.0.1", port)) + conn = s.accept().unwrap() + r = s.recv(conn, 0) + self.assertTrue(r.is_err()) + self.assertIn("must be positive", r.error.message) + finally: + client.close() + s.stop() + + def test_restrict_to_yields_an_unbound_capability(self): + # Socket state does not survive attenuation: a narrowed cap is + # something you take BEFORE you listen, not a second handle on + # an already-open listener. + s = Serve() + self.assertTrue(s.listen("127.0.0.1", 0).is_ok()) + try: + narrowed = s.restrict_to("127.0.0.1:*") + self.assertTrue(narrowed.local_port().is_err()) + self.assertTrue(narrowed.accept().is_err()) + finally: + s.stop() + + +# --------------------------------------------------------------------------- +# REGISTRIES AND THE TYPE SURFACE +# --------------------------------------------------------------------------- + + +class TestServeIsRegistered(unittest.TestCase): + """``Serve`` landed in every registry. The six guards in + tests/test_cap_handles.py are the general cross-check; these pin + the Serve-specific facts.""" + + def test_serve_is_a_known_capability_on_both_sides(self): + self.assertIn("Serve", CAPABILITY_NAMES) + self.assertIn("Serve", BUILTIN_CAPS) + + def test_serve_is_python_only_and_therefore_erased(self): + self.assertIn("Serve", PYTHON_ONLY_CAPS) + self.assertIn("Serve", ERASED_CAPS) + self.assertNotIn("Serve", HANDLE_BEARING_CAPS) + + def test_method_table_matches_the_runtime_class(self): + # A method the checker accepts but the runtime lacks is an + # AttributeError at run time, in a capability whose whole job + # is to be a trust boundary. + declared = {name for name, _ty, _tp in METHODS["Serve"]} + self.assertEqual( + declared, + {"restrict_to", "allows", "listen", "local_port", "accept", + "recv", "send", "close", "stop"}, + ) + for name in declared: + self.assertTrue( + callable(getattr(Serve, name, None)), + f"capa.runtime Serve has no method {name!r}", + ) + + def test_runtime_exports_serve(self): + import capa.runtime as rt + self.assertIn("Serve", rt.__all__) + self.assertIs(rt.Serve, Serve) + + def test_lsp_offers_serve(self): + from capa.lsp.completion import _BUILTIN_CAPABILITIES + self.assertIn("Serve", _BUILTIN_CAPABILITIES) + + def test_serve_send_is_an_ifc_sink_and_recv_is_not_a_secret_source(self): + from capa.analyzer._ifc import _PUBLIC_SINKS, _SECRET_SOURCES + # Sending bytes to a client is exfiltration, like Net.post. + # Argument 1 is the payload; argument 0 is the connection id. + self.assertEqual(_PUBLIC_SINKS[("Serve", "send")], {1}) + # Inbound data is @public: untrusted, but not confidential. + # Labelling it @secret would make echoing a request back to + # its own sender a violation, which is the normal case. + self.assertNotIn(("Serve", "recv"), _SECRET_SOURCES) + self.assertNotIn(("Serve", "accept"), _SECRET_SOURCES) + + def test_sink_method_names_are_unique_per_capability(self): + # THE INVARIANT ``_ifc_summary`` DEPENDS ON. Its cross-function + # summary pass has no receiver type available, so it attributes + # a built-in sink to a capability BY METHOD NAME alone. That is + # sound only while each sink method name belongs to exactly one + # capability. + # + # Serve was first drafted with a ``write``, which collided with + # ``Fs.write`` and silently made every ``fs.write`` report Serve + # as a reached capability -- a precision regression in a + # security analysis, found by + # tests/test_unaudited_secret_sink_fact.py. Renaming to ``send`` + # fixed it; this guard means the next capability cannot + # reintroduce the collision quietly. + from capa.analyzer._ifc import _PUBLIC_SINKS + seen: dict = {} + collisions = [] + for cap, method in _PUBLIC_SINKS: + if method in seen: + collisions.append((method, sorted([seen[method], cap]))) + seen[method] = cap + self.assertEqual( + collisions, [], + "two capabilities share a sink method name, so " + "_ifc_summary's by-method-name attribution would tag both " + f"for either one's calls: {collisions}", + ) + + def test_secret_source_method_names_are_unique_per_capability(self): + # Same argument, for the source side: ``_ifc_summary`` matches + # ``_SECRET_SOURCE_METHODS`` by name too. + from capa.analyzer._ifc import _SECRET_SOURCES + methods = [m for _c, m in _SECRET_SOURCES] + self.assertEqual(len(methods), len(set(methods))) + + +class TestServeTypeChecks(unittest.TestCase): + + def _check(self, source: str): + return _parse_and_analyze(source) + + def test_serve_annotation_is_accepted(self): + self._check( + "fun main(serve: Serve)\n" + " let s = serve.restrict_to(\"127.0.0.1:8080\")\n" + " let _ = s.listen(\"127.0.0.1\", 8080)\n" + ) + + def test_serve_methods_are_capability_gated(self): + # A function without a Serve parameter cannot listen: this is + # the whole point of the capability. + with self.assertRaises(Exception): + self._check( + "fun sneaky() -> Unit\n" + " let s = Serve()\n" + " let _ = s.listen(\"127.0.0.1\", 80)\n" + " return\n" + ) + + def test_demo_example_type_checks(self): + source = (_EXAMPLES / "serve_demo.capa").read_text(encoding="utf-8") + self._check(source) + + +# --------------------------------------------------------------------------- +# WASM REJECTION (no wasmtime required -- rejection happens at emit time) +# --------------------------------------------------------------------------- + + +class TestWasmRejectsServe(unittest.TestCase): + """The Wasm backend refuses ``Serve`` up front, following the + ``Unsafe`` precedent exactly: one early raise naming EVERY + offending site, saying why, and saying what to do instead. + + Deliberately free of wasmtime: ``emit_wat`` is pure text + generation and the rejection fires before anything is executed, so + this runs on the CI job that has no wasm extra installed.""" + + def _emit(self, module): + # Local import: the emitter is pure Python (no wasmtime), but + # keeping it out of the module-level import surface makes it + # obvious at a glance that this file collects on a runner with + # no wasm extra installed. + from capa.ir import emit_wat + return emit_wat(module) + + def _emit_source(self, source: str): + module, result = _parse_and_analyze(source) + return self._emit(lower(module, types=result.types)) + + def _error(self): + from capa.ir._emit_wasm._layout import WasmEmissionError + return WasmEmissionError + + def test_serve_parameter_is_rejected_with_a_clear_message(self): + source = (_EXAMPLES / "serve_demo.capa").read_text(encoding="utf-8") + with self.assertRaises(self._error()) as ctx: + self._emit_source(source) + msg = str(ctx.exception) + self.assertIn("Serve", msg) + # WHY. + self.assertIn("wasi:sockets", msg) + # WHAT TO DO INSTEAD. + self.assertIn("Python backend", msg) + # WHERE. + self.assertIn("main(serve: Serve)", msg) + + def test_every_offending_site_is_listed(self): + source = ( + "fun one(a: Serve) -> Unit\n" + " let _ = a.listen(\"127.0.0.1\", 8080)\n" + " return\n" + "fun two(b: Serve) -> Unit\n" + " let _ = b.listen(\"127.0.0.1\", 8081)\n" + " return\n" + "fun main(serve: Serve)\n" + " one(serve)\n" + " two(serve)\n" + ) + with self.assertRaises(self._error()) as ctx: + self._emit_source(source) + msg = str(ctx.exception) + for site in ("one(a: Serve)", "two(b: Serve)", "main(serve: Serve)"): + self.assertIn(site, msg) + + def test_serve_inside_a_struct_field_is_rejected(self): + # Reachability, not a literal head: the same recursive walk + # that closed the Unsafe hole (audit 2026-06-17 C5(b)). + from capa.ir._nodes import ( + Module, Function, Param, StructDecl, StructField, + ) + module = Module( + functions=[ + Function( + name="f", params=[Param(name="w", ty="Wrapper")], + return_type="Unit", declared_caps=[], body=[], + ), + ], + types=[ + StructDecl( + name="Wrapper", + fields=[StructField(name="s", ty="Serve")], + ), + ], + ) + with self.assertRaises(self._error()) as ctx: + self._emit(module) + msg = str(ctx.exception) + self.assertIn("Serve", msg) + self.assertIn("f(w: Wrapper)", msg) + + def test_serve_in_a_generic_argument_is_rejected(self): + from capa.ir._nodes import Module, Function, Param + module = Module( + functions=[ + Function( + name="f", params=[Param(name="xs", ty="List")], + return_type="Unit", declared_caps=[], body=[], + ), + ], + ) + with self.assertRaises(self._error()) as ctx: + self._emit(module) + self.assertIn("Serve", str(ctx.exception)) + + def test_unsafe_rejection_still_names_unsafe_not_serve(self): + # The scan is per-capability: generalising it must not blur the + # two diagnostics into one unhelpful message. + from capa.ir._nodes import Module, Function, Param + module = Module( + functions=[ + Function( + name="f", params=[Param(name="u", ty="Unsafe")], + return_type="Unit", declared_caps=[], body=[], + ), + ], + ) + with self.assertRaises(self._error()) as ctx: + self._emit(module) + msg = str(ctx.exception) + self.assertIn("Unsafe", msg) + self.assertIn("FFI", msg) + self.assertNotIn("Serve", msg) + + def test_a_serve_free_program_still_emits(self): + wat = self._emit_source( + "fun main(stdio: Stdio)\n stdio.println(\"hi\")\n" + ) + self.assertIn("(module", wat) + + +class TestWitRejectsPythonOnlyCaps(unittest.TestCase): + """``capa --wit`` is a STANDALONE path: it never runs the Wasm + emitter, so the discovery-time rejection does not fire for it. + + Before this was closed, ``--wit`` on a Serve program exited 0 and + printed a document that silently omitted Serve -- and whose + ``world`` block declared ``export main: func()`` while the real + ``main`` took a ``Serve`` parameter. A document whose entire + purpose is to describe the program's interface to a host was + describing a different program. + + Both members of ``PYTHON_ONLY_CAPS`` are covered, because they + reach the generator by DIFFERENT routes and only a + signature-level scan catches both: ``Serve`` has methods and so + lands in ``used``; ``Unsafe`` is method-less and never does (its + authority goes through the ``py_import`` / ``py_invoke`` free + functions), which is why the omission for Unsafe was invisible. + """ + + def _wit(self, source: str): + from capa.ir import emit_wit + module, result = _parse_and_analyze(source) + return emit_wit(lower(module, types=result.types)) + + def _error(self): + from capa.ir._emit_wit import PythonOnlyCapabilityInWit + return PythonOnlyCapabilityInWit + + def test_wit_rejects_serve(self): + source = (_EXAMPLES / "serve_demo.capa").read_text(encoding="utf-8") + with self.assertRaises(self._error()) as ctx: + self._wit(source) + msg = str(ctx.exception) + self.assertIn("Serve", msg) + self.assertIn("wasi:sockets", msg) + self.assertIn("main(serve: Serve)", msg) + self.assertEqual(ctx.exception.cap, "Serve") + + def test_wit_rejects_method_less_unsafe(self): + # The regression that a ``used``-based check could never catch: + # Unsafe has no capability methods, so it is invisible to + # ``collect_used_capabilities``. Only the signature scan sees it. + with self.assertRaises(self._error()) as ctx: + self._wit( + "fun main(stdio: Stdio, u: Unsafe)\n" + " let _m = py_import(u, \"math\")\n" + " stdio.println(\"x\")\n" + ) + self.assertIn("Unsafe", str(ctx.exception)) + self.assertEqual(ctx.exception.cap, "Unsafe") + + def test_wit_still_emits_for_a_normal_program(self): + wit = self._wit( + "fun main(stdio: Stdio, fs: Fs)\n" + " let _ = fs.read(\"x.txt\")\n" + " stdio.println(\"x\")\n" + ) + self.assertIn("interface stdio {", wit) + self.assertIn("interface fs {", wit) + + def test_cli_wit_exits_nonzero_on_serve(self): + # End to end through the real CLI, because the silent-success + # exit 0 was the actual user-visible bug. + import subprocess + import sys + proc = subprocess.run( + [sys.executable, "-m", "capa", "--wit", + str(_EXAMPLES / "serve_demo.capa")], + capture_output=True, text=True, + cwd=str(_EXAMPLES.parent.parent), + ) + self.assertEqual(proc.returncode, 1, proc.stdout) + self.assertIn("Serve", proc.stderr) + # And it must NOT have printed a WIT document. + self.assertNotIn("package capa:host", proc.stdout) + + +# --------------------------------------------------------------------------- +# MANIFEST / SBOM / POLICY (derived layers -- verified, not assumed) +# --------------------------------------------------------------------------- + + +_TWO_FUNCTION_SOURCE = '''\ +pub fun handles(serve: Serve) -> Unit + let _ = serve.listen("127.0.0.1", 8080) + return + +pub fun pure_helper(n: Int) -> Int + return n + 1 + +pub fun logs(stdio: Stdio) -> Unit + stdio.println("no serve here") + return +''' + + +class TestServeInTheManifest(unittest.TestCase): + + def _manifest(self, source: str, filename: str = "m.capa"): + module, result = _parse_and_analyze(source, filename) + return build_manifest( + module, filename=filename, expr_labels=result.expr_labels, + unaudited_secret_sinks=result.unaudited_secret_sinks, + ) + + def test_serve_appears_on_exactly_the_functions_that_hold_it(self): + manifest = self._manifest(_TWO_FUNCTION_SOURCE) + by_name = {f["name"]: f for f in manifest["functions"]} + + self.assertIn("Serve", by_name["handles"]["declared_capabilities"]) + self.assertIn( + "Serve", + by_name["handles"]["transitively_reachable_capabilities"], + ) + + # And on nothing else. A capability that leaks into unrelated + # functions would make the manifest useless as an audit + # artefact. + for name in ("pure_helper", "logs"): + self.assertNotIn( + "Serve", by_name[name]["declared_capabilities"], name, + ) + self.assertNotIn( + "Serve", + by_name[name]["transitively_reachable_capabilities"], + name, + ) + + def test_serve_propagates_transitively_but_not_into_declared(self): + manifest = self._manifest( + "fun inner(serve: Serve) -> Unit\n" + " let _ = serve.listen(\"127.0.0.1\", 80)\n" + " return\n" + "fun outer(serve: Serve) -> Unit\n" + " inner(serve)\n" + " return\n" + ) + by_name = {f["name"]: f for f in manifest["functions"]} + self.assertIn( + "Serve", by_name["outer"]["transitively_reachable_capabilities"], + ) + + def test_restrict_to_is_recorded_as_an_attenuation(self): + # ``restrict_to`` keeps its exact name so the existing + # attenuation tracking in capa/manifest/_flow.py works on Serve + # with no change. This confirms it does. + manifest = self._manifest( + "fun run(serve: Serve) -> Unit\n" + " let narrow = serve.restrict_to(\"127.0.0.1:8080\")\n" + " let _ = narrow.listen(\"127.0.0.1\", 8080)\n" + " return\n" + ) + blob = str(manifest) + self.assertIn("restrict_to", blob) + self.assertIn("127.0.0.1:8080", blob) + + +class TestServeInSbomAndPolicy(unittest.TestCase): + """``Serve`` reaches the composed SBOM and a ``forbid-capability`` + policy over it fires. Both layers derive from + ``CAPABILITY_NAMES``; this is the verification of that, not an + assumption about it.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.tmp = Path(self._tmp.name) + self.addCleanup(self._tmp.cleanup) + + def _write(self, rel: str, text: str) -> None: + p = self.tmp / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(text, encoding="utf-8") + + def _app(self, main_src: str, policy: str) -> Path: + root = self.tmp / "app" + self._write( + "app/capa.toml", + '[package]\nname = "app"\nversion = "0.1.0"\n', + ) + self._write("app/main.capa", main_src) + self._write("app/capa-policy.toml", policy) + return root + + def _compose(self, root: Path): + root = root.resolve() + filename = str(root / "main.capa") + source = Path(filename).read_text(encoding="utf-8") + loader = ModuleLoader(search_paths=[root]) + linked = loader.load_root(source, filename) + result = analyze( + linked.module, source=source, filename=filename, + sources=linked.sources, module_privates=linked.module_privates, + ) + if not result.ok: + raise AssertionError(f"analyzer errors: {result.errors}") + manifest = build_manifest( + linked.module, filename=filename, + expr_labels=result.expr_labels, + unaudited_secret_sinks=result.unaudited_secret_sinks, + ) + return build_composed_sbom(linked.module, manifest, root) + + def _evaluate(self, root: Path, policy_id: str): + composed = self._compose(root) + policy_path = find_policy_file(root.resolve()) + policies = read_policy_file(policy_path) if policy_path else [] + report = evaluate_policies(composed, policies) + for r in report["results"]: + if r["policy"] == policy_id: + return r + raise AssertionError(f"no policy result {policy_id!r}") + + def test_serve_appears_in_the_composed_sbom(self): + root = self._app( + "pub fun run(serve: Serve) -> Unit\n" + " let _ = serve.listen(\"127.0.0.1\", 8080)\n" + " return\n", + "", + ) + composed = self._compose(root) + self.assertIn("Serve", str(composed)) + + def test_forbid_capability_policy_over_serve_fires(self): + root = self._app( + "pub fun run(serve: Serve) -> Unit\n" + " let _ = serve.listen(\"127.0.0.1\", 8080)\n" + " return\n", + '[[policy]]\nid = "no-serve"\nkind = "forbid-capability"\n' + 'capability = "Serve"\n', + ) + result = self._evaluate(root, "no-serve") + self.assertFalse(result["pass"], "the forbid-capability did not fire") + self.assertEqual(result["violations"][0]["capability"], "Serve") + + def test_forbid_capability_policy_passes_on_a_serve_free_product(self): + root = self._app( + "pub fun run(fs: Fs) -> Unit\n" + " let _ = fs.read(\"x.txt\")\n" + " return\n", + '[[policy]]\nid = "no-serve"\nkind = "forbid-capability"\n' + 'capability = "Serve"\n', + ) + self.assertTrue(self._evaluate(root, "no-serve")["pass"]) + + +if __name__ == "__main__": + unittest.main()