diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml new file mode 100644 index 0000000..79ce591 --- /dev/null +++ b/.github/workflows/lint.yml @@ -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 . \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index d10f872..5b05e3a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,8 @@ dependencies = [ [project.optional-dependencies] dev = [ "pytest>=7.0", - "pip-audit", + "black>=24.0", + "ruff>=0.6", ] [tool.mypy] @@ -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" diff --git a/stellargate/adapters/rytscan.py b/stellargate/adapters/rytscan.py index 9ecbe2f..165f320 100644 --- a/stellargate/adapters/rytscan.py +++ b/stellargate/adapters/rytscan.py @@ -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 @@ -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})" diff --git a/stellargate/adapters/schemalock.py b/stellargate/adapters/schemalock.py index 065db03..1e7826e 100644 --- a/stellargate/adapters/schemalock.py +++ b/stellargate/adapters/schemalock.py @@ -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 @@ -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" @@ -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 "" diff --git a/stellargate/adapters/shieldscan.py b/stellargate/adapters/shieldscan.py index 6a42e3d..ce5ed86 100644 --- a/stellargate/adapters/shieldscan.py +++ b/stellargate/adapters/shieldscan.py @@ -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 diff --git a/stellargate/adapters/vaultsweep.py b/stellargate/adapters/vaultsweep.py index 597c31f..31eb327 100644 --- a/stellargate/adapters/vaultsweep.py +++ b/stellargate/adapters/vaultsweep.py @@ -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 @@ -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: diff --git a/stellargate/aggregator.py b/stellargate/aggregator.py index 42371f3..000ab5a 100644 --- a/stellargate/aggregator.py +++ b/stellargate/aggregator.py @@ -1,4 +1,5 @@ """Runs enabled adapters and collects all findings.""" + from __future__ import annotations from stellargate.adapters import rytscan, schemalock, shieldscan, vaultsweep @@ -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 diff --git a/stellargate/cli.py b/stellargate/cli.py index ca49bfe..d26d7a3 100644 --- a/stellargate/cli.py +++ b/stellargate/cli.py @@ -1,4 +1,5 @@ """stellargate run --config stellargate.yaml [--json-report path] [--fail-on LEVEL]""" + from __future__ import annotations import argparse @@ -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) diff --git a/stellargate/config.py b/stellargate/config.py index fb08903..748a6fe 100644 --- a/stellargate/config.py +++ b/stellargate/config.py @@ -1,4 +1,5 @@ """Load and validate stellargate.yaml.""" + from __future__ import annotations from dataclasses import dataclass, field @@ -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}") @@ -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] = {} diff --git a/stellargate/report.py b/stellargate/report.py index e32127a..a5c19eb 100644 --- a/stellargate/report.py +++ b/stellargate/report.py @@ -1,4 +1,5 @@ """JSON and Markdown compliance report generation.""" + from __future__ import annotations from datetime import datetime, timezone @@ -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.") diff --git a/stellargate/schema.py b/stellargate/schema.py index f0ac0f8..871ee60 100644 --- a/stellargate/schema.py +++ b/stellargate/schema.py @@ -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} @@ -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: diff --git a/tests/test_adapters.py b/tests/test_adapters.py index a15ee83..b1cc336 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -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(): diff --git a/tests/test_aggregator_and_report.py b/tests/test_aggregator_and_report.py index 46c194b..3568137 100644 --- a/tests/test_aggregator_and_report.py +++ b/tests/test_aggregator_and_report.py @@ -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"