From f3ef394e31355ff4ae8fbc93ed48e7b3b5a21a44 Mon Sep 17 00:00:00 2001 From: Prekzursil Date: Fri, 26 Jun 2026 02:34:44 +0300 Subject: [PATCH] chore: purge retired QZP SaaS support code + add workflow least-privilege MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes the support code for the retired Quality Zero (QZP) / visual-SaaS machinery (workflows already deleted in #91): - scripts/quality/*.py — QZP SaaS gate scripts (check_codacy_zero, check_sonar_zero, check_sentry_zero, check_deepscan_zero, check_required_checks, check_quality_secrets, assert_coverage_100); only ever invoked by the deleted quality-zero workflows. Clears the CodeQL py/path-injection + py/partial-ssrf + unused-variable alerts rooted in these files. - frontend/webcoder_ui/e2e/** + playwright.chromatic.config.ts and the orphaned e2e:chromatic / e2e:applitools npm scripts — Chromatic / Applitools visual-SaaS tests, run only by the retired visual workflows. Also adds 'permissions: contents: read' to django.yml and django_ci.yml (least privilege), clearing the CodeQL actions/missing-workflow-permissions alerts. The lean 'quality' gate and 'verify' do not reference any removed path. --- .github/workflows/django.yml | 3 + .github/workflows/django_ci.yml | 3 + .../e2e/applitools-core-routes.spec.ts | 72 ------ .../e2e/chromatic/public-routes.spec.ts | 33 --- frontend/webcoder_ui/e2e/support/publicApp.ts | 55 ---- frontend/webcoder_ui/package.json | 4 +- .../playwright.chromatic.config.ts | 27 -- scripts/quality/assert_coverage_100.py | 212 ---------------- scripts/quality/check_codacy_zero.py | 226 ----------------- scripts/quality/check_deepscan_zero.py | 175 ------------- scripts/quality/check_quality_secrets.py | 159 ------------ scripts/quality/check_required_checks.py | 216 ---------------- scripts/quality/check_sentry_zero.py | 205 --------------- scripts/quality/check_sonar_zero.py | 234 ------------------ 14 files changed, 7 insertions(+), 1617 deletions(-) delete mode 100644 frontend/webcoder_ui/e2e/applitools-core-routes.spec.ts delete mode 100644 frontend/webcoder_ui/e2e/chromatic/public-routes.spec.ts delete mode 100644 frontend/webcoder_ui/e2e/support/publicApp.ts delete mode 100644 frontend/webcoder_ui/playwright.chromatic.config.ts delete mode 100644 scripts/quality/assert_coverage_100.py delete mode 100644 scripts/quality/check_codacy_zero.py delete mode 100644 scripts/quality/check_deepscan_zero.py delete mode 100644 scripts/quality/check_quality_secrets.py delete mode 100644 scripts/quality/check_required_checks.py delete mode 100644 scripts/quality/check_sentry_zero.py delete mode 100755 scripts/quality/check_sonar_zero.py diff --git a/.github/workflows/django.yml b/.github/workflows/django.yml index 840a21a4..3409bcf6 100644 --- a/.github/workflows/django.yml +++ b/.github/workflows/django.yml @@ -6,6 +6,9 @@ on: pull_request: branches: ["main", "master"] +permissions: + contents: read + jobs: build: runs-on: ubuntu-latest diff --git a/.github/workflows/django_ci.yml b/.github/workflows/django_ci.yml index 9311d263..96514403 100644 --- a/.github/workflows/django_ci.yml +++ b/.github/workflows/django_ci.yml @@ -10,6 +10,9 @@ on: paths: - 'backend/**' +permissions: + contents: read + jobs: build: runs-on: ubuntu-latest diff --git a/frontend/webcoder_ui/e2e/applitools-core-routes.spec.ts b/frontend/webcoder_ui/e2e/applitools-core-routes.spec.ts deleted file mode 100644 index 12c801ce..00000000 --- a/frontend/webcoder_ui/e2e/applitools-core-routes.spec.ts +++ /dev/null @@ -1,72 +0,0 @@ -import fs from 'node:fs'; - -import { BatchInfo, Configuration, Eyes, Target } from '@applitools/eyes-playwright'; -import { test } from '@playwright/test'; - -import { preparePublicVisualState, stabilizeRoute } from './support/publicApp'; - -function numberOrZero(value: unknown): number { - if (typeof value === 'number' && Number.isFinite(value)) return value; - if (typeof value === 'string' && value.trim() !== '' && !Number.isNaN(Number(value))) - return Number(value); - return 0; -} - -function buildConfiguration(): Configuration { - const configuration = new Configuration(); - configuration.setApiKey(process.env.APPLITOOLS_API_KEY || ''); - configuration.setAppName('WebCoder'); - configuration.setBatch( - new BatchInfo( - process.env.APPLITOOLS_BATCH_NAME || `WebCoder-${process.env.GITHUB_SHA || 'local'}`, - ), - ); - configuration.setMatchLevel('Strict'); - return configuration; -} - -const routes = [ - { name: 'Home', path: '/' }, - { name: 'Login', path: '/login' }, - { name: 'Register', path: '/register' }, - { name: 'Problems', path: '/problems' }, -]; - -test('capture public routes with Applitools', async ({ page }) => { - test.skip(!process.env.APPLITOOLS_API_KEY, 'APPLITOOLS_API_KEY is required'); - - await preparePublicVisualState(page); - - const resultsPath = process.env.APPLITOOLS_RESULTS_PATH || 'applitools/results.json'; - const eyes = new Eyes(); - eyes.setConfiguration(buildConfiguration()); - - await eyes.open(page, 'WebCoder', 'public-routes', { width: 1366, height: 900 }); - - try { - for (const route of routes) { - await stabilizeRoute(page, route.path); - await eyes.check(route.name, Target.window().fully()); - } - - const closeResult = await eyes.close(); - const payload = { - unresolved: numberOrZero( - (closeResult as any)?.getUnresolved?.() ?? (closeResult as any)?.unresolved, - ), - mismatches: numberOrZero( - (closeResult as any)?.getMismatches?.() ?? (closeResult as any)?.mismatches, - ), - missing: numberOrZero((closeResult as any)?.getMissing?.() ?? (closeResult as any)?.missing), - }; - - fs.mkdirSync('applitools', { recursive: true }); - fs.writeFileSync(resultsPath, JSON.stringify(payload, null, 2)); - - if (payload.unresolved || payload.mismatches || payload.missing) { - throw new Error(`Applitools visual diff detected: ${JSON.stringify(payload)}`); - } - } finally { - await eyes.abortIfNotClosed(); - } -}); diff --git a/frontend/webcoder_ui/e2e/chromatic/public-routes.spec.ts b/frontend/webcoder_ui/e2e/chromatic/public-routes.spec.ts deleted file mode 100644 index f734f9f0..00000000 --- a/frontend/webcoder_ui/e2e/chromatic/public-routes.spec.ts +++ /dev/null @@ -1,33 +0,0 @@ -import { expect, test } from '@chromatic-com/playwright'; - -import { preparePublicVisualState, stabilizeRoute } from '../support/publicApp'; - -test.beforeEach(async ({ page }) => { - await preparePublicVisualState(page); -}); - -test('home route renders', async ({ page }) => { - await stabilizeRoute(page, '/'); - await expect(page.getByRole('heading', { name: /welcome to webcoder/i })).toBeVisible(); - await expect(page.getByRole('link', { name: /view problems/i })).toBeVisible(); - await expect(page.getByRole('link', { name: /sign up/i })).toBeVisible(); -}); - -test('login route renders', async ({ page }) => { - await stabilizeRoute(page, '/login'); - await expect(page.getByRole('heading', { name: /login/i })).toBeVisible(); - await expect(page.getByRole('link', { name: /^google$/i })).toBeVisible(); - await expect(page.getByRole('link', { name: /^github$/i })).toBeVisible(); -}); - -test('register route renders', async ({ page }) => { - await stabilizeRoute(page, '/register'); - await expect(page.getByRole('heading', { name: /register/i })).toBeVisible(); - await expect(page.getByRole('button', { name: /register/i })).toBeVisible(); -}); - -test('problems route renders', async ({ page }) => { - await stabilizeRoute(page, '/problems'); - await expect(page.getByRole('heading', { name: /^problems$/i })).toBeVisible(); - await expect(page.getByRole('link', { name: /two sum warmup/i })).toBeVisible(); -}); diff --git a/frontend/webcoder_ui/e2e/support/publicApp.ts b/frontend/webcoder_ui/e2e/support/publicApp.ts deleted file mode 100644 index 4e8346a0..00000000 --- a/frontend/webcoder_ui/e2e/support/publicApp.ts +++ /dev/null @@ -1,55 +0,0 @@ -import type { Page, Route } from '@playwright/test'; - -const sampleProblems = [ - { - id: 101, - title_i18n: { - en: 'Two Sum Warmup', - }, - difficulty: 'EASY', - status: 'PUBLISHED', - }, -]; - -async function fulfillJson(route: Route, body: unknown, status = 200): Promise { - await route.fulfill({ - status, - contentType: 'application/json', - body: JSON.stringify(body), - }); -} - -export async function preparePublicVisualState(page: Page): Promise { - await page.addInitScript(() => { - localStorage.removeItem('accessToken'); - localStorage.removeItem('refreshToken'); - localStorage.removeItem('user'); - }); - - await page.route('http://127.0.0.1:8000/api/v1/**', async (route) => { - if (route.request().url().includes('/problems/problems/')) { - await fulfillJson(route, { data: sampleProblems }); - return; - } - if (route.request().method() === 'GET') { - await fulfillJson(route, []); - return; - } - await fulfillJson(route, { detail: 'Visual test request was mocked.' }); - }); -} - -export async function stabilizeRoute(page: Page, path: string): Promise { - await page.goto(path, { waitUntil: 'domcontentloaded' }); - await page.addStyleTag({ - content: ` - *, *::before, *::after { - animation-duration: 0s !important; - animation-delay: 0s !important; - transition: none !important; - caret-color: transparent !important; - } - `, - }); - await page.waitForTimeout(250); -} diff --git a/frontend/webcoder_ui/package.json b/frontend/webcoder_ui/package.json index 8aedddb5..4b5dec8f 100644 --- a/frontend/webcoder_ui/package.json +++ b/frontend/webcoder_ui/package.json @@ -30,9 +30,7 @@ "build": "react-scripts build", "test": "react-scripts test", "test:coverage": "react-scripts test --coverage --watchAll=false --ci", - "eject": "react-scripts eject", - "e2e:chromatic": "playwright test --config ./playwright.chromatic.config.ts ./e2e/chromatic/public-routes.spec.ts", - "e2e:applitools": "playwright test --config ./playwright.chromatic.config.ts ./e2e/applitools-core-routes.spec.ts --project=chromium --workers=1" + "eject": "react-scripts eject" }, "eslintConfig": { "extends": [ diff --git a/frontend/webcoder_ui/playwright.chromatic.config.ts b/frontend/webcoder_ui/playwright.chromatic.config.ts deleted file mode 100644 index eb5e506d..00000000 --- a/frontend/webcoder_ui/playwright.chromatic.config.ts +++ /dev/null @@ -1,27 +0,0 @@ -import { defineConfig, devices } from '@playwright/test'; - -export default defineConfig({ - testDir: './e2e', - timeout: 60_000, - expect: { timeout: 10_000 }, - retries: process.env.CI ? 1 : 0, - reporter: [['line']], - use: { - baseURL: process.env.E2E_BASE_URL || 'http://127.0.0.1:3100', - trace: 'on-first-retry', - screenshot: 'only-on-failure', - video: 'retain-on-failure', - }, - webServer: { - command: 'BROWSER=none HOST=127.0.0.1 PORT=3100 CI=true npm start', - url: 'http://127.0.0.1:3100', - reuseExistingServer: !process.env.CI, - timeout: 900_000, - }, - projects: [ - { - name: 'chromium', - use: { ...devices['Desktop Chrome'] }, - }, - ], -}); diff --git a/scripts/quality/assert_coverage_100.py b/scripts/quality/assert_coverage_100.py deleted file mode 100644 index 0831285c..00000000 --- a/scripts/quality/assert_coverage_100.py +++ /dev/null @@ -1,212 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import json -import re -import sys -from dataclasses import dataclass -from datetime import datetime, timezone -from pathlib import Path - - -@dataclass -class CoverageStats: - name: str - path: str - covered: int - total: int - - @property - def percent(self) -> float: - if self.total <= 0: - return 100.0 - return (self.covered / self.total) * 100.0 - - -_PAIR_RE = re.compile(r"^(?P[^=]+)=(?P.+)$") -_XML_LINES_VALID_RE = re.compile(r'lines-valid="([0-9]+(?:\\.[0-9]+)?)"') -_XML_LINES_COVERED_RE = re.compile(r'lines-covered="([0-9]+(?:\\.[0-9]+)?)"') -_XML_LINE_HITS_RE = re.compile(r"]*\\bhits=\"([0-9]+(?:\\.[0-9]+)?)\"") - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Assert 100% coverage for all declared components." - ) - parser.add_argument( - "--xml", action="append", default=[], help="Coverage XML input: name=path" - ) - parser.add_argument( - "--lcov", action="append", default=[], help="LCOV input: name=path" - ) - parser.add_argument( - "--out-json", default="coverage-100/coverage.json", help="Output JSON path" - ) - parser.add_argument( - "--out-md", default="coverage-100/coverage.md", help="Output markdown path" - ) - return parser.parse_args() - - -def parse_named_path(value: str) -> tuple[str, Path]: - match = _PAIR_RE.match(value.strip()) - if not match: - raise ValueError(f"Invalid input '{value}'. Expected format: name=path") - return match.group("name").strip(), Path(match.group("path").strip()) - - -def parse_coverage_xml(name: str, path: Path) -> CoverageStats: - text = path.read_text(encoding="utf-8") - lines_valid_match = _XML_LINES_VALID_RE.search(text) - lines_covered_match = _XML_LINES_COVERED_RE.search(text) - - if lines_valid_match and lines_covered_match: - total = int(float(lines_valid_match.group(1))) - covered = int(float(lines_covered_match.group(1))) - return CoverageStats(name=name, path=str(path), covered=covered, total=total) - - total = 0 - covered = 0 - for hits_raw in _XML_LINE_HITS_RE.findall(text): - total += 1 - try: - if int(float(hits_raw)) > 0: - covered += 1 - except ValueError: - continue - - return CoverageStats(name=name, path=str(path), covered=covered, total=total) - - -def parse_lcov(name: str, path: Path) -> CoverageStats: - total = 0 - covered = 0 - - for raw in path.read_text(encoding="utf-8").splitlines(): - line = raw.strip() - if line.startswith("LF:"): - total += int(line.split(":", 1)[1]) - elif line.startswith("LH:"): - covered += int(line.split(":", 1)[1]) - - return CoverageStats(name=name, path=str(path), covered=covered, total=total) - - -def evaluate(stats: list[CoverageStats]) -> tuple[str, list[str]]: - findings: list[str] = [] - for item in stats: - if item.percent < 100.0: - findings.append( - f"{item.name} coverage below 100%: {item.percent:.2f}% ({item.covered}/{item.total})" - ) - - combined_total = sum(item.total for item in stats) - combined_covered = sum(item.covered for item in stats) - combined = ( - 100.0 if combined_total <= 0 else (combined_covered / combined_total) * 100.0 - ) - - if combined < 100.0: - findings.append( - f"combined coverage below 100%: {combined:.2f}% ({combined_covered}/{combined_total})" - ) - - status = "pass" if not findings else "fail" - return status, findings - - -def _render_md(payload: dict) -> str: - lines = [ - "# Coverage 100 Gate", - "", - f"- Status: `{payload['status']}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Components", - ] - - for item in payload.get("components", []): - lines.append( - f"- `{item['name']}`: `{item['percent']:.2f}%` ({item['covered']}/{item['total']}) from `{item['path']}`" - ) - - if not payload.get("components"): - lines.append("- None") - - lines.extend(["", "## Findings"]) - findings = payload.get("findings") or [] - if findings: - lines.extend(f"- {finding}" for finding in findings) - else: - lines.append("- None") - - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def main() -> int: - args = _parse_args() - - stats: list[CoverageStats] = [] - for item in args.xml: - name, path = parse_named_path(item) - stats.append(parse_coverage_xml(name, path)) - for item in args.lcov: - name, path = parse_named_path(item) - stats.append(parse_lcov(name, path)) - - if not stats: - raise SystemExit( - "No coverage files were provided; pass --xml and/or --lcov inputs." - ) - - status, findings = evaluate(stats) - payload = { - "status": status, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - "components": [ - { - "name": item.name, - "path": item.path, - "covered": item.covered, - "total": item.total, - "percent": item.percent, - } - for item in stats - ], - "findings": findings, - } - - try: - out_json = _safe_output_path(args.out_json, "coverage-100/coverage.json") - out_md = _safe_output_path(args.out_md, "coverage-100/coverage.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - out_json.write_text( - json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - - return 0 if status == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/quality/check_codacy_zero.py b/scripts/quality/check_codacy_zero.py deleted file mode 100644 index 3090f09d..00000000 --- a/scripts/quality/check_codacy_zero.py +++ /dev/null @@ -1,226 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import json -import sys -import urllib.error -import urllib.parse -import urllib.request -from datetime import datetime, timezone -from pathlib import Path -from typing import Any - -_SCRIPT_DIR = Path(__file__).resolve().parent -_HELPER_ROOT = ( - _SCRIPT_DIR - if (_SCRIPT_DIR / "security_helpers.py").exists() - else _SCRIPT_DIR.parent -) -if str(_HELPER_ROOT) not in sys.path: - sys.path.insert(0, str(_HELPER_ROOT)) - -from security_helpers import normalize_https_url # noqa: E402 (import requires the sys.path insert above) - - -TOTAL_KEYS = {"total", "totalItems", "total_items", "count", "hits", "open_issues"} -CODACY_API_BASE = "https://api.codacy.com" - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Assert Codacy has zero total open issues." - ) - parser.add_argument( - "--provider", default="gh", help="Organization provider, for example gh" - ) - parser.add_argument("--owner", required=True, help="Repository owner") - parser.add_argument("--repo", required=True, help="Repository name") - parser.add_argument( - "--token", - default="", - help="Codacy API token (falls back to CODACY_API_TOKEN env)", - ) - parser.add_argument( - "--out-json", default="codacy-zero/codacy.json", help="Output JSON path" - ) - parser.add_argument( - "--out-md", default="codacy-zero/codacy.md", help="Output markdown path" - ) - return parser.parse_args() - - -def _request_json( - url: str, token: str, *, method: str = "GET", data: dict[str, Any] | None = None -) -> dict[str, Any]: - safe_url = normalize_https_url(url, allowed_host_suffixes={"codacy.com"}).rstrip( - "/" - ) - body = None - headers = { - "Accept": "application/json", - "api-token": token, - "User-Agent": "reframe-codacy-zero-gate", - } - if data is not None: - body = json.dumps(data).encode("utf-8") - headers["Content-Type"] = "application/json" - req = urllib.request.Request( - safe_url, - headers=headers, - method=method, - data=body, - ) - with urllib.request.urlopen(req, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) - - -def extract_total_open(payload: Any) -> int | None: - if isinstance(payload, dict): - for key, value in payload.items(): - if key in TOTAL_KEYS and isinstance(value, (int, float)): - return int(value) - - # common pagination structures - for key in ("pagination", "page", "meta"): - nested = payload.get(key) - total = extract_total_open(nested) - if total is not None: - return total - - for value in payload.values(): - total = extract_total_open(value) - if total is not None: - return total - - if isinstance(payload, list): - for item in payload: - total = extract_total_open(item) - if total is not None: - return total - - return None - - -def _render_md(payload: dict) -> str: - lines = [ - "# Codacy Zero Gate", - "", - f"- Status: `{payload['status']}`", - f"- Owner/repo: `{payload['owner']}/{payload['repo']}`", - f"- Open issues: `{payload.get('open_issues')}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Findings", - ] - findings = payload.get("findings") or [] - if findings: - lines.extend(f"- {item}" for item in findings) - else: - lines.append("- None") - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def main() -> int: - import os - - args = _parse_args() - token = (args.token or os.environ.get("CODACY_API_TOKEN", "")).strip() - api_base = normalize_https_url( - CODACY_API_BASE, allowed_hosts={"api.codacy.com"} - ).rstrip("/") - owner = urllib.parse.quote(args.owner.strip(), safe="") - repo = urllib.parse.quote(args.repo.strip(), safe="") - - findings: list[str] = [] - open_issues: int | None = None - - if not token: - findings.append("CODACY_API_TOKEN is missing.") - status = "fail" - else: - query = urllib.parse.urlencode({"limit": "1"}) - provider_candidates = [args.provider, "gh", "github"] - provider_candidates = list(dict.fromkeys(p for p in provider_candidates if p)) - - last_exc: Exception | None = None - for provider in provider_candidates: - url = ( - f"{api_base}/api/v3/analysis/organizations/{provider}/" - f"{owner}/repositories/{repo}/issues/search?{query}" - ) - try: - payload = _request_json(url, token, method="POST", data={}) - open_issues = extract_total_open(payload) - if open_issues is None: - findings.append( - "Codacy response did not include a parseable total issue count." - ) - elif open_issues != 0: - findings.append( - f"Codacy reports {open_issues} open issues (expected 0)." - ) - status = "pass" if not findings else "fail" - break - except urllib.error.HTTPError as exc: - last_exc = exc - if exc.code == 404: - continue - findings.append(f"Codacy API request failed: HTTP {exc.code}") - status = "fail" - break - except Exception as exc: # pragma: no cover - network/runtime surface - last_exc = exc - findings.append(f"Codacy API request failed: {exc}") - status = "fail" - break - else: - findings.append( - f"Codacy API endpoint was not found for provider(s): {', '.join(provider_candidates)}." - ) - if last_exc is not None: - findings.append(f"Last Codacy API error: {last_exc}") - status = "fail" - - payload = { - "status": status, - "owner": args.owner, - "repo": args.repo, - "provider": args.provider, - "open_issues": open_issues, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - "findings": findings, - } - - try: - out_json = _safe_output_path(args.out_json, "codacy-zero/codacy.json") - out_md = _safe_output_path(args.out_md, "codacy-zero/codacy.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - out_json.write_text( - json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - return 0 if status == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/quality/check_deepscan_zero.py b/scripts/quality/check_deepscan_zero.py deleted file mode 100644 index c9416dec..00000000 --- a/scripts/quality/check_deepscan_zero.py +++ /dev/null @@ -1,175 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import json -import sys -import urllib.request -from datetime import datetime, timezone -from pathlib import Path -from typing import Any - -_SCRIPT_DIR = Path(__file__).resolve().parent -_HELPER_ROOT = ( - _SCRIPT_DIR - if (_SCRIPT_DIR / "security_helpers.py").exists() - else _SCRIPT_DIR.parent -) -if str(_HELPER_ROOT) not in sys.path: - sys.path.insert(0, str(_HELPER_ROOT)) - -from security_helpers import normalize_https_url # noqa: E402 (import requires the sys.path insert above) - -TOTAL_KEYS = {"total", "totalItems", "total_items", "count", "hits", "open_issues"} - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Assert DeepScan has zero total open issues." - ) - parser.add_argument( - "--token", - default="", - help="DeepScan API token (falls back to DEEPSCAN_API_TOKEN env)", - ) - parser.add_argument( - "--out-json", default="deepscan-zero/deepscan.json", help="Output JSON path" - ) - parser.add_argument( - "--out-md", default="deepscan-zero/deepscan.md", help="Output markdown path" - ) - return parser.parse_args() - - -def extract_total_open(payload: Any) -> int | None: - if isinstance(payload, dict): - for key, value in payload.items(): - if key in TOTAL_KEYS and isinstance(value, (int, float)): - return int(value) - for nested in payload.values(): - total = extract_total_open(nested) - if total is not None: - return total - elif isinstance(payload, list): - for nested in payload: - total = extract_total_open(nested) - if total is not None: - return total - return None - - -def _request_json(url: str, token: str) -> dict[str, Any]: - safe_url = normalize_https_url(url, allowed_host_suffixes={"deepscan.io"}) - req = urllib.request.Request( - safe_url, - headers={ - "Accept": "application/json", - "Authorization": f"Bearer {token}", - "User-Agent": "reframe-deepscan-zero-gate", - }, - method="GET", - ) - with urllib.request.urlopen(req, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) - - -def _render_md(payload: dict) -> str: - lines = [ - "# DeepScan Zero Gate", - "", - f"- Status: `{payload['status']}`", - f"- Open issues: `{payload.get('open_issues')}`", - f"- Source URL: `{payload.get('open_issues_url') or 'n/a'}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Findings", - ] - findings = payload.get("findings") or [] - if findings: - lines.extend(f"- {item}" for item in findings) - else: - lines.append("- None") - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def main() -> int: - import os - - args = _parse_args() - token = (args.token or os.environ.get("DEEPSCAN_API_TOKEN", "")).strip() - open_issues_url = os.environ.get("DEEPSCAN_OPEN_ISSUES_URL", "").strip() - - findings: list[str] = [] - open_issues: int | None = None - - if not token: - findings.append("DEEPSCAN_API_TOKEN is missing.") - if not open_issues_url: - findings.append("DEEPSCAN_OPEN_ISSUES_URL is missing.") - else: - try: - open_issues_url = normalize_https_url( - open_issues_url, - allowed_host_suffixes={"deepscan.io"}, - ) - except ValueError as exc: - findings.append(str(exc)) - - status = "fail" - if not findings: - try: - payload = _request_json(open_issues_url, token) - open_issues = extract_total_open(payload) - if open_issues is None: - findings.append( - "DeepScan response did not include a parseable total issue count." - ) - elif open_issues != 0: - findings.append( - f"DeepScan reports {open_issues} open issues (expected 0)." - ) - status = "pass" if not findings else "fail" - except Exception as exc: # pragma: no cover - network/runtime surface - findings.append(f"DeepScan API request failed: {exc}") - status = "fail" - - payload = { - "status": status, - "open_issues": open_issues, - "open_issues_url": open_issues_url, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - "findings": findings, - } - - try: - out_json = _safe_output_path(args.out_json, "deepscan-zero/deepscan.json") - out_md = _safe_output_path(args.out_md, "deepscan-zero/deepscan.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - out_json.write_text( - json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - return 0 if status == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/quality/check_quality_secrets.py b/scripts/quality/check_quality_secrets.py deleted file mode 100644 index 52e18c80..00000000 --- a/scripts/quality/check_quality_secrets.py +++ /dev/null @@ -1,159 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import json -import os -import sys -from datetime import datetime, timezone -from pathlib import Path - -DEFAULT_REQUIRED_SECRETS = [ - "SONAR_TOKEN", - "CODACY_API_TOKEN", - "SENTRY_AUTH_TOKEN", - "APPLITOOLS_API_KEY", -] - -DEFAULT_REQUIRED_VARS = [ - "SENTRY_ORG", - "SENTRY_PROJECT", -] - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Validate required quality-gate secrets/variables are configured." - ) - parser.add_argument( - "--required-secret", - action="append", - default=[], - help="Additional required secret env var name", - ) - parser.add_argument( - "--required-var", - action="append", - default=[], - help="Additional required variable env var name", - ) - parser.add_argument( - "--out-json", default="quality-secrets/secrets.json", help="Output JSON path" - ) - parser.add_argument( - "--out-md", default="quality-secrets/secrets.md", help="Output markdown path" - ) - return parser.parse_args() - - -def _dedupe(items: list[str]) -> list[str]: - seen: set[str] = set() - out: list[str] = [] - for item in items: - key = str(item or "").strip() - if not key or key in seen: - continue - seen.add(key) - out.append(key) - return out - - -def evaluate_env( - required_secrets: list[str], required_vars: list[str] -) -> dict[str, list[str]]: - missing_secrets = [ - name for name in required_secrets if not str(os.environ.get(name, "")).strip() - ] - missing_vars = [ - name for name in required_vars if not str(os.environ.get(name, "")).strip() - ] - present_secrets = [name for name in required_secrets if name not in missing_secrets] - present_vars = [name for name in required_vars if name not in missing_vars] - return { - "missing_secrets": missing_secrets, - "missing_vars": missing_vars, - "present_secrets": present_secrets, - "present_vars": present_vars, - } - - -def _render_md(payload: dict) -> str: - lines = [ - "# Quality Secrets Preflight", - "", - f"- Status: `{payload['status']}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Missing secrets", - ] - missing_secrets = payload.get("missing_secrets") or [] - if missing_secrets: - lines.extend(f"- `{name}`" for name in missing_secrets) - else: - lines.append("- None") - - lines.extend(["", "## Missing variables"]) - missing_vars = payload.get("missing_vars") or [] - if missing_vars: - lines.extend(f"- `{name}`" for name in missing_vars) - else: - lines.append("- None") - - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def main() -> int: - args = _parse_args() - required_secrets = _dedupe( - DEFAULT_REQUIRED_SECRETS + list(args.required_secret or []) - ) - required_vars = _dedupe(DEFAULT_REQUIRED_VARS + list(args.required_var or [])) - - result = evaluate_env(required_secrets, required_vars) - status = ( - "pass" - if not result["missing_secrets"] and not result["missing_vars"] - else "fail" - ) - payload = { - "status": status, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - "required_secrets": required_secrets, - "required_vars": required_vars, - **result, - } - - try: - out_json = _safe_output_path(args.out_json, "quality-secrets/secrets.json") - out_md = _safe_output_path(args.out_md, "quality-secrets/secrets.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - - out_json.write_text( - json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - - return 0 if status == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/quality/check_required_checks.py b/scripts/quality/check_required_checks.py deleted file mode 100644 index 29bcd0d9..00000000 --- a/scripts/quality/check_required_checks.py +++ /dev/null @@ -1,216 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import json -import os -import sys -import time -import urllib.parse -import urllib.request -from datetime import datetime, timezone -from pathlib import Path -from typing import Any - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Wait for required GitHub check contexts and assert they are successful." - ) - parser.add_argument("--repo", required=True, help="owner/repo") - parser.add_argument("--sha", required=True, help="commit SHA") - parser.add_argument( - "--required-context", action="append", default=[], help="Required context name" - ) - parser.add_argument("--timeout-seconds", type=int, default=900) - parser.add_argument("--poll-seconds", type=int, default=20) - parser.add_argument("--out-json", default="quality-zero-gate/required-checks.json") - parser.add_argument("--out-md", default="quality-zero-gate/required-checks.md") - return parser.parse_args() - - -def _api_get(repo: str, path: str, token: str) -> dict[str, Any]: - url = f"https://api.github.com/repos/{repo}/{path.lstrip('/')}" - req = urllib.request.Request( - url, - headers={ - "Accept": "application/vnd.github+json", - "Authorization": f"Bearer {token}", - "X-GitHub-Api-Version": "2022-11-28", - "User-Agent": "reframe-quality-zero-gate", - }, - method="GET", - ) - with urllib.request.urlopen(req, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) - - -def _collect_contexts( - check_runs_payload: dict[str, Any], status_payload: dict[str, Any] -) -> dict[str, dict[str, str]]: - contexts: dict[str, dict[str, str]] = {} - - for run in check_runs_payload.get("check_runs", []) or []: - name = str(run.get("name") or "").strip() - if not name: - continue - contexts[name] = { - "state": str(run.get("status") or ""), - "conclusion": str(run.get("conclusion") or ""), - "source": "check_run", - } - - for status in status_payload.get("statuses", []) or []: - name = str(status.get("context") or "").strip() - if not name: - continue - contexts[name] = { - "state": str(status.get("state") or ""), - "conclusion": str(status.get("state") or ""), - "source": "status", - } - - return contexts - - -def _evaluate( - required: list[str], contexts: dict[str, dict[str, str]] -) -> tuple[str, list[str], list[str]]: - missing: list[str] = [] - failed: list[str] = [] - - for context in required: - observed = contexts.get(context) - if not observed: - missing.append(context) - continue - - source = observed.get("source") - if source == "check_run": - state = observed.get("state") - conclusion = observed.get("conclusion") - if state != "completed": - failed.append(f"{context}: status={state}") - elif conclusion != "success": - failed.append(f"{context}: conclusion={conclusion}") - else: - conclusion = observed.get("conclusion") - if conclusion != "success": - failed.append(f"{context}: state={conclusion}") - - status = "pass" if not missing and not failed else "fail" - return status, missing, failed - - -def _render_md(payload: dict) -> str: - lines = [ - "# Quality Zero Gate - Required Contexts", - "", - f"- Status: `{payload['status']}`", - f"- Repo/SHA: `{payload['repo']}@{payload['sha']}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Missing contexts", - ] - - missing = payload.get("missing") or [] - if missing: - lines.extend(f"- `{name}`" for name in missing) - else: - lines.append("- None") - - lines.extend(["", "## Failed contexts"]) - failed = payload.get("failed") or [] - if failed: - lines.extend(f"- {entry}" for entry in failed) - else: - lines.append("- None") - - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def main() -> int: - args = _parse_args() - token = ( - os.environ.get("GITHUB_TOKEN", "") or os.environ.get("GH_TOKEN", "") - ).strip() - required = [item.strip() for item in args.required_context if item.strip()] - - if not required: - raise SystemExit("At least one --required-context is required") - if not token: - raise SystemExit("GITHUB_TOKEN or GH_TOKEN is required") - - deadline = time.time() + max(args.timeout_seconds, 1) - - final_payload: dict[str, Any] | None = None - while time.time() <= deadline: - check_runs = _api_get( - args.repo, f"commits/{args.sha}/check-runs?per_page=100", token - ) - statuses = _api_get(args.repo, f"commits/{args.sha}/status", token) - contexts = _collect_contexts(check_runs, statuses) - status, missing, failed = _evaluate(required, contexts) - - final_payload = { - "status": status, - "repo": args.repo, - "sha": args.sha, - "required": required, - "missing": missing, - "failed": failed, - "contexts": contexts, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - } - - if status == "pass": - break - - # wait only while there are missing contexts or in-progress check-runs - in_progress = any( - v.get("state") != "completed" - for v in contexts.values() - if v.get("source") == "check_run" - ) - if not missing and not in_progress: - break - time.sleep(max(args.poll_seconds, 1)) - - if final_payload is None: - raise SystemExit("No payload collected") - - try: - out_json = _safe_output_path( - args.out_json, "quality-zero-gate/required-checks.json" - ) - out_md = _safe_output_path(args.out_md, "quality-zero-gate/required-checks.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - out_json.write_text( - json.dumps(final_payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(final_payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - - return 0 if final_payload["status"] == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/quality/check_sentry_zero.py b/scripts/quality/check_sentry_zero.py deleted file mode 100644 index f5c7a31b..00000000 --- a/scripts/quality/check_sentry_zero.py +++ /dev/null @@ -1,205 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import json -import sys -import urllib.parse -import urllib.request -from datetime import datetime, timezone -from pathlib import Path -from typing import Any - -_SCRIPT_DIR = Path(__file__).resolve().parent -_HELPER_ROOT = ( - _SCRIPT_DIR - if (_SCRIPT_DIR / "security_helpers.py").exists() - else _SCRIPT_DIR.parent -) -if str(_HELPER_ROOT) not in sys.path: - sys.path.insert(0, str(_HELPER_ROOT)) - -from security_helpers import normalize_https_url # noqa: E402 (import requires the sys.path insert above) - -SENTRY_API_BASE = "https://sentry.io/api/0" - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Assert Sentry has zero unresolved issues for configured projects." - ) - parser.add_argument( - "--org", default="", help="Sentry org slug (falls back to SENTRY_ORG env)" - ) - parser.add_argument( - "--project", - action="append", - default=[], - help="Project slug (repeatable, falls back to SENTRY_PROJECT_BACKEND/SENTRY_PROJECT_WEB env)", - ) - parser.add_argument( - "--token", - default="", - help="Sentry auth token (falls back to SENTRY_AUTH_TOKEN env)", - ) - parser.add_argument( - "--out-json", default="sentry-zero/sentry.json", help="Output JSON path" - ) - parser.add_argument( - "--out-md", default="sentry-zero/sentry.md", help="Output markdown path" - ) - return parser.parse_args() - - -def _request(url: str, token: str) -> tuple[list[Any], dict[str, str]]: - safe_url = normalize_https_url(url, allowed_host_suffixes={"sentry.io"}) - req = urllib.request.Request( - safe_url, - headers={ - "Accept": "application/json", - "Authorization": f"Bearer {token}", - "User-Agent": "reframe-sentry-zero-gate", - }, - method="GET", - ) - with urllib.request.urlopen(req, timeout=30) as resp: - body = json.loads(resp.read().decode("utf-8")) - headers = {k.lower(): v for k, v in resp.headers.items()} - if not isinstance(body, list): - raise RuntimeError("Unexpected Sentry response payload") - return body, headers - - -def _hits_from_headers(headers: dict[str, str]) -> int | None: - raw = headers.get("x-hits") - if not raw: - return None - try: - return int(raw) - except ValueError: - return None - - -def _render_md(payload: dict) -> str: - lines = [ - "# Sentry Zero Gate", - "", - f"- Status: `{payload['status']}`", - f"- Org: `{payload.get('org')}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Project results", - ] - - for item in payload.get("projects", []): - lines.append(f"- `{item['project']}` unresolved=`{item['unresolved']}`") - - if not payload.get("projects"): - lines.append("- None") - - lines.extend(["", "## Findings"]) - findings = payload.get("findings") or [] - if findings: - lines.extend(f"- {item}" for item in findings) - else: - lines.append("- None") - - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def main() -> int: - import os - - args = _parse_args() - token = (args.token or os.environ.get("SENTRY_AUTH_TOKEN", "")).strip() - org = (args.org or os.environ.get("SENTRY_ORG", "")).strip() - api_base = normalize_https_url(SENTRY_API_BASE, allowed_hosts={"sentry.io"}).rstrip( - "/" - ) - - projects = [p for p in args.project if p] - if not projects: - for env_name in ("SENTRY_PROJECT_BACKEND", "SENTRY_PROJECT_WEB"): - value = str(os.environ.get(env_name, "")).strip() - if value: - projects.append(value) - - findings: list[str] = [] - project_results: list[dict[str, Any]] = [] - - if not token: - findings.append("SENTRY_AUTH_TOKEN is missing.") - if not org: - findings.append("SENTRY_ORG is missing.") - if not projects: - findings.append( - "No Sentry projects configured (SENTRY_PROJECT_BACKEND/SENTRY_PROJECT_WEB)." - ) - - status = "fail" - if not findings: - try: - for project in projects: - query = urllib.parse.urlencode({"query": "is:unresolved", "limit": "1"}) - org_slug = urllib.parse.quote(org, safe="") - project_slug = urllib.parse.quote(project, safe="") - url = f"{api_base}/projects/{org_slug}/{project_slug}/issues/?{query}" - issues, headers = _request(url, token) - unresolved = _hits_from_headers(headers) - if unresolved is None: - unresolved = len(issues) - if unresolved >= 1: - findings.append( - f"Sentry project {project} returned unresolved issues but no X-Hits header for exact totals." - ) - if unresolved != 0: - findings.append( - f"Sentry project {project} has {unresolved} unresolved issues (expected 0)." - ) - project_results.append({"project": project, "unresolved": unresolved}) - - status = "pass" if not findings else "fail" - except Exception as exc: # pragma: no cover - network/runtime surface - findings.append(f"Sentry API request failed: {exc}") - status = "fail" - - payload = { - "status": status, - "org": org, - "projects": project_results, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - "findings": findings, - } - - try: - out_json = _safe_output_path(args.out_json, "sentry-zero/sentry.json") - out_md = _safe_output_path(args.out_md, "sentry-zero/sentry.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - out_json.write_text( - json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - return 0 if status == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/quality/check_sonar_zero.py b/scripts/quality/check_sonar_zero.py deleted file mode 100755 index 9a53b88a..00000000 --- a/scripts/quality/check_sonar_zero.py +++ /dev/null @@ -1,234 +0,0 @@ -#!/usr/bin/env python3 -from __future__ import annotations - -import argparse -import base64 -import json -import sys -import urllib.parse -import urllib.request -from datetime import datetime, timezone -from pathlib import Path -from typing import Any - -_SCRIPT_DIR = Path(__file__).resolve().parent -_HELPER_ROOT = ( - _SCRIPT_DIR - if (_SCRIPT_DIR / "security_helpers.py").exists() - else _SCRIPT_DIR.parent -) -if str(_HELPER_ROOT) not in sys.path: - sys.path.insert(0, str(_HELPER_ROOT)) - -from security_helpers import normalize_https_url # noqa: E402 (import requires the sys.path insert above) - -SONAR_API_BASE = "https://sonarcloud.io" -UNRESOLVED_HOTSPOT_STATUS = "TO_REVIEW" - - -def _parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description=( - "Assert SonarCloud has zero open issues, zero unresolved security hotspots, " - "and a passing quality gate." - ) - ) - parser.add_argument("--project-key", required=True, help="Sonar project key") - parser.add_argument( - "--token", default="", help="Sonar token (falls back to SONAR_TOKEN env)" - ) - parser.add_argument("--branch", default="", help="Optional branch scope") - parser.add_argument("--pull-request", default="", help="Optional PR scope") - parser.add_argument( - "--out-json", default="sonar-zero/sonar.json", help="Output JSON path" - ) - parser.add_argument( - "--out-md", default="sonar-zero/sonar.md", help="Output markdown path" - ) - return parser.parse_args() - - -def _auth_header(token: str) -> str: - raw = f"{token}:".encode("utf-8") - return "Basic " + base64.b64encode(raw).decode("ascii") - - -def _request_json(url: str, auth_header: str) -> dict[str, Any]: - safe_url = normalize_https_url(url, allowed_host_suffixes={"sonarcloud.io"}).rstrip( - "/" - ) - request = urllib.request.Request( - safe_url, - headers={ - "Accept": "application/json", - "Authorization": auth_header, - "User-Agent": "reframe-sonar-zero-gate", - }, - method="GET", - ) - with urllib.request.urlopen(request, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) - - -def _render_md(payload: dict) -> str: - lines = [ - "# Sonar Zero Gate", - "", - f"- Status: `{payload['status']}`", - f"- Project: `{payload['project_key']}`", - f"- Open issues: `{payload.get('open_issues')}`", - f"- Security hotspots total: `{payload.get('security_hotspots_total')}`", - f"- Security hotspots to review: `{payload.get('security_hotspots_to_review')}`", - f"- Quality gate: `{payload.get('quality_gate')}`", - f"- Timestamp (UTC): `{payload['timestamp_utc']}`", - "", - "## Findings", - ] - findings = payload.get("findings") or [] - if findings: - lines.extend(f"- {item}" for item in findings) - else: - lines.append("- None") - return "\n".join(lines) + "\n" - - -def _safe_output_path(raw: str, fallback: str, base: Path | None = None) -> Path: - root = (base or Path.cwd()).resolve() - candidate = Path((raw or "").strip() or fallback).expanduser() - if not candidate.is_absolute(): - candidate = root / candidate - resolved = candidate.resolve(strict=False) - try: - resolved.relative_to(root) - except ValueError as exc: - raise ValueError(f"Output path escapes workspace root: {candidate}") from exc - return resolved - - -def _scope_query(project_key: str, branch: str, pull_request: str) -> dict[str, str]: - query = {"projectKey": project_key} - if branch: - query["branch"] = branch - if pull_request: - query["pullRequest"] = pull_request - return query - - -def _search_total( - api_base: str, endpoint: str, query: dict[str, str], auth_header: str -) -> int: - url = f"{api_base}{endpoint}?{urllib.parse.urlencode(query)}" - payload = _request_json(url, auth_header) - paging = payload.get("paging") or {} - return int(paging.get("total") or 0) - - -def main() -> int: - import os - - args = _parse_args() - token = (args.token or os.environ.get("SONAR_TOKEN", "")).strip() - api_base = normalize_https_url( - SONAR_API_BASE, allowed_hosts={"sonarcloud.io"} - ).rstrip("/") - - findings: list[str] = [] - open_issues: int | None = None - quality_gate: str | None = None - security_hotspots_total: int | None = None - security_hotspots_to_review: int | None = None - - if not token: - findings.append("SONAR_TOKEN is missing.") - status = "fail" - else: - auth = _auth_header(token) - try: - issues_query = { - "componentKeys": args.project_key, - "resolved": "false", - "ps": "1", - } - if args.branch: - issues_query["branch"] = args.branch - if args.pull_request: - issues_query["pullRequest"] = args.pull_request - - open_issues = _search_total( - api_base, "/api/issues/search", issues_query, auth - ) - - hotspots_query = _scope_query( - args.project_key, args.branch, args.pull_request - ) - hotspots_query["ps"] = "1" - security_hotspots_total = _search_total( - api_base, - "/api/hotspots/search", - dict(hotspots_query), - auth, - ) - to_review_query = dict(hotspots_query) - to_review_query["status"] = UNRESOLVED_HOTSPOT_STATUS - security_hotspots_to_review = _search_total( - api_base, - "/api/hotspots/search", - to_review_query, - auth, - ) - - gate_query = _scope_query(args.project_key, args.branch, args.pull_request) - gate_url = f"{api_base}/api/qualitygates/project_status?{urllib.parse.urlencode(gate_query)}" - gate_payload = _request_json(gate_url, auth) - project_status = gate_payload.get("projectStatus") or {} - quality_gate = str(project_status.get("status") or "UNKNOWN") - - if open_issues != 0: - findings.append( - f"Sonar reports {open_issues} open issues (expected 0)." - ) - if security_hotspots_to_review != 0: - findings.append( - f"Sonar reports {security_hotspots_to_review} unresolved security hotspots (expected 0)." - ) - if quality_gate != "OK": - findings.append( - f"Sonar quality gate status is {quality_gate} (expected OK)." - ) - - status = "pass" if not findings else "fail" - except Exception as exc: # pragma: no cover - network/runtime surface - status = "fail" - findings.append(f"Sonar API request failed: {exc}") - - payload = { - "status": status, - "project_key": args.project_key, - "open_issues": open_issues, - "security_hotspots_total": security_hotspots_total, - "security_hotspots_to_review": security_hotspots_to_review, - "quality_gate": quality_gate, - "timestamp_utc": datetime.now(timezone.utc).isoformat(), - "findings": findings, - } - - try: - out_json = _safe_output_path(args.out_json, "sonar-zero/sonar.json") - out_md = _safe_output_path(args.out_md, "sonar-zero/sonar.md") - except ValueError as exc: - print(str(exc), file=sys.stderr) - return 1 - - out_json.parent.mkdir(parents=True, exist_ok=True) - out_md.parent.mkdir(parents=True, exist_ok=True) - out_json.write_text( - json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" - ) - out_md.write_text(_render_md(payload), encoding="utf-8") - print(out_md.read_text(encoding="utf-8"), end="") - - return 0 if status == "pass" else 1 - - -if __name__ == "__main__": - raise SystemExit(main())