Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions .github/workflows/lint.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
name: lint

on:
push:
pull_request:

jobs:
lint:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11"]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install ruff and black
run: |
python -m pip install --upgrade pip
pip install ruff black
- name: ruff check
run: ruff check .
- name: black check
run: black --check .
11 changes: 10 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@ dependencies = [
[project.optional-dependencies]
dev = [
"pytest>=7.0",
"pip-audit",
"black>=24.0",
"ruff>=0.6",
]

[tool.mypy]
Expand All @@ -21,6 +22,14 @@ check_untyped_defs = true
[project.scripts]
stellargate = "stellargate.cli:main"

[tool.ruff]
line-length = 100
target-version = "py310"

[tool.black]
line-length = 100
target-version = ["py310"]

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
14 changes: 11 additions & 3 deletions stellargate/adapters/rytscan.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
are passed straight through as our rule_id — no remapping needed, they're
already namespaced and human-readable.
"""

from __future__ import annotations

import hashlib
Expand Down Expand Up @@ -54,11 +55,18 @@ def run(options: dict) -> list[Finding]:

def _scan_cli(path: str) -> list[Finding]:
cmd = [
"cargo", "run", "-p", "rytscan-cli", "--",
"scan", path, "--format", "json",
"cargo",
"run",
"-p",
"rytscan-cli",
"--",
"scan",
path,
"--format",
"json",
]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300, check=False)
except FileNotFoundError as e:
raise AdapterError(
f"{TOOL_NAME}: 'cargo' not found — is the Rust toolchain installed? ({e})"
Expand Down
41 changes: 18 additions & 23 deletions stellargate/adapters/schemalock.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
since a passing contract check isn't something a reviewer needs to see in
a compliance report.
"""

from __future__ import annotations

import json
Expand All @@ -29,9 +30,9 @@
# SchemaLock doesn't emit its own severity per check; we map by failure
# type since an auth-bypass is categorically worse than a status-code drift.
FAILURE_SEVERITY = {
"auth_required": "critical", # silent auth bypass
"error_envelope": "medium", # response shape drift
"status": "high", # wrong status code (e.g. leaks existence)
"auth_required": "critical", # silent auth bypass
"error_envelope": "medium", # response shape drift
"status": "high", # wrong status code (e.g. leaks existence)
}
DEFAULT_SEVERITY = "medium"

Expand All @@ -45,27 +46,21 @@ def run(options: dict) -> list[Finding]:
with tempfile.TemporaryDirectory() as tmp:
report_path = Path(tmp) / "schemalock-report.json"
cmd = [
"schemalock", "test",
"--config", config_path,
"--base-url", base_url,
"--json-report", str(report_path),
"schemalock",
"test",
"--config",
config_path,
"--base-url",
base_url,
"--json-report",
str(report_path),
]
last_stderr = ""
for attempt in range(2):
try:
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=SCHEMALOCK_TIMEOUT_SECONDS)
except FileNotFoundError as e:
# Retrying cannot help a missing binary — fail fast.
raise AdapterError(f"{TOOL_NAME}: 'schemalock' CLI not found ({e})")
except subprocess.TimeoutExpired:
# Same for a hang — the report will never arrive.
raise AdapterError(f"{TOOL_NAME}: test run timed out after {SCHEMALOCK_TIMEOUT_SECONDS}s")

last_stderr = (proc.stderr or "").strip()
if proc.returncode == 0 and report_path.exists():
break
if attempt == 0:
time.sleep(RETRY_DELAY_SECONDS)
try:
subprocess.run(cmd, capture_output=True, text=True, timeout=120, check=False)
except FileNotFoundError as e:
raise AdapterError(f"{TOOL_NAME}: 'schemalock' CLI not found ({e})")
except subprocess.TimeoutExpired:
raise AdapterError(f"{TOOL_NAME}: test run timed out after 120s")

if not report_path.exists():
detail = f" (last run stderr: {last_stderr})" if last_stderr else ""
Expand Down
1 change: 1 addition & 0 deletions stellargate/adapters/shieldscan.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
without a KeyError, and so config validation clearly rejects `enabled: true`
until Phase 2 lands.
"""

from __future__ import annotations

from stellargate.schema import AdapterError, Finding
Expand Down
3 changes: 2 additions & 1 deletion stellargate/adapters/vaultsweep.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
Rule IDs (STELLAR-001, MNEMONIC-001, API-00x, DEFAULT-001, RPC-001) are
already well-namespaced and pass through unchanged.
"""

from __future__ import annotations

import fnmatch
Expand Down Expand Up @@ -39,7 +40,7 @@ def run(options: dict) -> list[Finding]:
path = options.get("path", ".")
cmd = ["vaultsweep", "scan", path, "--format", "json"]
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
result = subprocess.run(cmd, capture_output=True, text=True, timeout=180, check=False)
except FileNotFoundError as e:
raise AdapterError(f"{TOOL_NAME}: 'vaultsweep' CLI not found ({e})")
except subprocess.TimeoutExpired:
Expand Down
3 changes: 2 additions & 1 deletion stellargate/aggregator.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Runs enabled adapters and collects all findings."""

from __future__ import annotations

from stellargate.adapters import rytscan, schemalock, shieldscan, vaultsweep
Expand Down Expand Up @@ -33,7 +34,7 @@ def run_all(config: Config) -> list[ToolRunResult]:
results.append(ToolRunResult(name, findings, None))
except AdapterError as e:
results.append(ToolRunResult(name, None, str(e)))
except Exception as e:
except Exception as e: # noqa: BLE001 - a buggy adapter must surface, not crash the run
# An adapter bug or unforeseen tool-output shape must not take
# down the whole run — every other tool's findings still belong
# in the report. Surface this as a clearly-labeled tool error
Expand Down
5 changes: 4 additions & 1 deletion stellargate/cli.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""stellargate run --config stellargate.yaml [--json-report path] [--fail-on LEVEL]"""

from __future__ import annotations

import argparse
Expand Down Expand Up @@ -57,7 +58,9 @@ def main(argv: list[str] | None = None) -> int:
run_parser.add_argument("--config", default="stellargate.yaml")
run_parser.add_argument("--json-report", default=None, help="Write JSON report to this path")
run_parser.add_argument("--md-report", default=None, help="Write Markdown report to this path")
run_parser.add_argument("--fail-on", default=None, help="Override fail_on threshold from config")
run_parser.add_argument(
"--fail-on", default=None, help="Override fail_on threshold from config"
)

args = parser.parse_args(argv)

Expand Down
10 changes: 4 additions & 6 deletions stellargate/config.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Load and validate stellargate.yaml."""

from __future__ import annotations

from dataclasses import dataclass, field
Expand Down Expand Up @@ -28,7 +29,7 @@ class Config:
tools: dict[str, ToolConfig]

@staticmethod
def load(path: str | Path) -> "Config":
def load(path: str | Path) -> Config:
path = Path(path)
if not path.exists():
raise ConfigError(f"Config file not found: {path}")
Expand All @@ -42,16 +43,13 @@ def load(path: str | Path) -> "Config":
target = raw.get("target", ".")
fail_on = raw.get("fail_on", "high").lower()
if fail_on not in VALID_SEVERITIES:
raise ConfigError(
f"Invalid fail_on '{fail_on}'; must be one of {VALID_SEVERITIES}"
)
raise ConfigError(f"Invalid fail_on '{fail_on}'; must be one of {VALID_SEVERITIES}")

raw_tools = raw.get("tools", {})
unknown = set(raw_tools) - set(KNOWN_TOOLS)
if unknown:
raise ConfigError(
f"Unknown tool(s) in config: {sorted(unknown)}; "
f"known tools are {KNOWN_TOOLS}"
f"Unknown tool(s) in config: {sorted(unknown)}; " f"known tools are {KNOWN_TOOLS}"
)

tools: dict[str, ToolConfig] = {}
Expand Down
5 changes: 2 additions & 3 deletions stellargate/report.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""JSON and Markdown compliance report generation."""

from __future__ import annotations

from datetime import datetime, timezone
Expand Down Expand Up @@ -61,9 +62,7 @@ def to_markdown(results: list[ToolRunResult], fail_on: str, gate_passed: bool) -
lines.append("|---|---|---|---|---|")
for f in findings:
loc = f.location or "—"
lines.append(
f"| {f.severity.upper()} | {f.tool} | {f.rule_id} | {loc} | {f.message} |"
)
lines.append(f"| {f.severity.upper()} | {f.tool} | {f.rule_id} | {loc} | {f.message} |")
else:
lines.append("No findings. Clean run.")

Expand Down
7 changes: 4 additions & 3 deletions stellargate/schema.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
"""Unified finding schema every adapter normalizes into."""

from __future__ import annotations

from dataclasses import dataclass, field, asdict
from typing import Any, Optional
from dataclasses import asdict, dataclass, field
from typing import Any

SEVERITY_ORDER = {"critical": 3, "high": 2, "medium": 1, "low": 0}

Expand All @@ -13,7 +14,7 @@ class Finding:
rule_id: str
severity: str
message: str
location: Optional[str] = None
location: str | None = None
raw: dict[str, Any] = field(default_factory=dict)

def __post_init__(self) -> None:
Expand Down
29 changes: 22 additions & 7 deletions tests/test_adapters.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,23 +113,38 @@ def __init__(self, stdout="", returncode=0, stderr=""):
def test_rytscan_crash_with_no_output_raises_not_zero_findings():
"""Regression test: a tool that crashes (nonzero exit, empty stdout) must
raise AdapterError, never be silently read as a clean 'zero findings' scan."""
with patch("subprocess.run", return_value=_FakeCompletedProcess(stdout="", returncode=1, stderr="panic: build failed")):
with pytest.raises(AdapterError, match="scan failed"):
rytscan.run({"path": "./contracts"})
with (
patch(
"subprocess.run",
return_value=_FakeCompletedProcess(
stdout="", returncode=1, stderr="panic: build failed"
),
),
pytest.raises(AdapterError, match="scan failed"),
):
rytscan.run({"path": "./contracts"})


def test_rytscan_clean_pass_with_zero_findings_is_fine():
"""A genuine clean scan (exit 0, empty JSON findings list) is a legitimate pass —
only a crash (nonzero exit + empty stdout) should raise."""
with patch("subprocess.run", return_value=_FakeCompletedProcess(stdout='{"findings": []}', returncode=0)):
with patch(
"subprocess.run",
return_value=_FakeCompletedProcess(stdout='{"findings": []}', returncode=0),
):
findings = rytscan.run({"path": "./contracts"})
assert findings == []


def test_vaultsweep_crash_with_no_output_raises_not_zero_findings():
with patch("subprocess.run", return_value=_FakeCompletedProcess(stdout="", returncode=2, stderr="permission denied")):
with pytest.raises(AdapterError, match="scan failed"):
vaultsweep.run({"path": "."})
with (
patch(
"subprocess.run",
return_value=_FakeCompletedProcess(stdout="", returncode=2, stderr="permission denied"),
),
pytest.raises(AdapterError, match="scan failed"),
):
vaultsweep.run({"path": "."})


def test_schemalock_mixed_report_maps_all_failed_checks():
Expand Down
2 changes: 1 addition & 1 deletion tests/test_aggregator_and_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ def test_to_json_shape():
assert len(report["findings"]) == 2
tool_names = {t["tool"] for t in report["tools"]}
assert tool_names == {"rytscan", "vaultsweep", "schemalock"}
errored = [t for t in report["tools"] if t["tool"] == "schemalock"][0]
errored = next(t for t in report["tools"] if t["tool"] == "schemalock")
assert errored["error"] == "schemalock CLI not found"


Expand Down
Loading