fix(protocol): harden gateway contract validation
Verify Protocol / verify (push) Canceled after 0s
Verify Protocol / module (push) Successful in 2m12s

This commit is contained in:
sechmachine
2026-07-29 21:16:53 +07:00
parent 0ea21cd3f2
commit ebfe07376d
18 changed files with 337 additions and 44 deletions
+28 -15
View File
@@ -30,6 +30,9 @@ SECRET_PATTERNS = (
re.compile(rb"\bgh[pousr]_[A-Za-z0-9]{20,}\b"),
re.compile(rb"\bsk-[A-Za-z0-9]{20,}\b"),
)
ALLOWED_SECRET_PROPERTIES = {
("ProviderSessionWork", "client_private_key_pem"),
}
def fail(message: str) -> None:
@@ -53,19 +56,36 @@ def check_generated_provenance() -> None:
def check_manifest_schema() -> None:
schema = json.loads((ROOT / "schemas/control-v1.schema.json").read_text(encoding="utf-8"))
definitions = schema.get("$defs", {})
for name in ("ConnectionManifest", "ManifestGateway", "ManifestTunnel", "ManifestProfile", "ManifestBounds", "GrantReference"):
properties = definitions.get(name, {}).get("properties", {})
forbidden = sorted(FORBIDDEN_WIRE_FIELDS.intersection(properties))
if forbidden:
fail(f"{name} exposes forbidden wire fields: {forbidden}")
for name, definition in definitions.items():
for field in definition.get("properties", {}):
if any(forbidden in field.lower() for forbidden in FORBIDDEN_WIRE_FIELDS):
if (name, field) not in ALLOWED_SECRET_PROPERTIES:
fail(f"{name} exposes forbidden wire field {field}")
def check_proto_boundaries(path: pathlib.Path) -> None:
message = ""
depth = 0
for line in path.read_text(encoding="utf-8").splitlines():
match = re.match(r"\s*message\s+([A-Za-z0-9_]+)\s*\{", line)
if match and depth == 0:
message = match.group(1)
if any(field in line.lower() for field in FORBIDDEN_WIRE_FIELDS):
allowed = (
message == "ProviderSessionWork"
and re.fullmatch(r"\s*string\s+client_private_key_pem\s*=\s*[0-9]+;\s*", line)
)
if not allowed:
fail(f"{path.relative_to(ROOT)} exposes forbidden wire field in {message or 'file scope'}")
depth += line.count("{") - line.count("}")
if depth == 0:
message = ""
def check_text_boundaries() -> None:
paths = [
ROOT / "openapi/control-v1.yaml",
ROOT / "schemas/control-v1.schema.json",
ROOT / "proto/versevdi/control/v1/control.proto",
ROOT / "proto/versevdi/tunnel/v1/tunnel.proto",
ROOT / "frames/datagram-v1.md",
ROOT / "frames/registry.json",
ROOT / "registries/features.json",
@@ -77,14 +97,7 @@ def check_text_boundaries() -> None:
if field in text:
fail(f"{path.relative_to(ROOT)} contains forbidden wire field {field}")
generated_paths = list((ROOT / "gen").rglob("*"))
for path in generated_paths:
if not path.is_file() or path.name == "manifest.json" or path.suffix in {".pb", ".binpb"}:
continue
text = path.read_text(encoding="utf-8").lower()
for field in FORBIDDEN_WIRE_FIELDS:
if field in text:
fail(f"generated output {path.relative_to(ROOT)} contains forbidden wire field {field}")
check_proto_boundaries(ROOT / "proto/versevdi/tunnel/v1/tunnel.proto")
def check_secret_canaries() -> None: