diff --git a/lib/hypatia/cli.ex b/lib/hypatia/cli.ex index cf8ea6a0..f9cb8e03 100644 --- a/lib/hypatia/cli.ex +++ b/lib/hypatia/cli.ex @@ -17,6 +17,8 @@ defmodule Hypatia.CLI do hypatia pr-eligibility --owner X Query AM010 admin-merge eligibility for a PR --repo Y --pr N + hypatia secret-alerts --owner X Secret-scanning alerts for one repo, as one JSON line + --repo Y (exit 3 if they cannot be fetched) hypatia version Print version and exit hypatia help Print usage and exit @@ -155,6 +157,9 @@ defmodule Hypatia.CLI do ["pr-eligibility" | _] -> run_pr_eligibility(opts[:owner], opts[:repo], opts[:pr]) + ["secret-alerts" | _] -> + run_secret_alerts(opts[:owner], opts[:repo]) + ["version"] -> IO.puts("hypatia #{@version}") @@ -284,6 +289,50 @@ defmodule Hypatia.CLI do System.halt(2) end + # ─── Secret-scanning alerts for one repo (estate sweep) ────────────── + + @doc false + def run_secret_alerts(owner, repo) when is_binary(owner) and is_binary(repo) do + {line, exit_code} = secret_alerts_output(owner, repo) + IO.puts(line) + if exit_code != 0, do: System.halt(exit_code) + end + + def run_secret_alerts(_, _) do + IO.puts(:stderr, "Error: secret-alerts requires --owner and --repo.") + IO.puts(:stderr, "Usage: hypatia secret-alerts --owner OWNER --repo REPO") + System.halt(2) + end + + @doc false + # One JSON line, and the exit code to return: 0 when the alerts were read + # (with or without findings), 3 when they could not be. A sweep must be + # able to tell "no alerts" from "could not look", so a refused token or a + # repo with secret scanning disabled is never reported as clean. + def secret_alerts_output( + owner, + repo, + sweep \\ &Hypatia.Rules.SecretScanningAlerts.sweep/2 + ) do + case sweep.(owner, repo) do + {:ok, result} -> + line = + Jason.encode!(%{ + owner: owner, + repo: repo, + status: "ok", + total: result.total, + by_severity: result.by_severity, + findings: result.findings + }) + + {line, 0} + + {:error, reason} -> + {Jason.encode!(%{owner: owner, repo: repo, status: "error", error: reason}), 3} + end + end + # ─── Diagnostic summary ────────────────────────────────────────────── # # Always emit a single-line summary on stderr after a scan/report so CI @@ -1578,6 +1627,11 @@ defmodule Hypatia.CLI do --owner OWNER GitHub owner (org or user) --repo REPO Repository name --pr NUMBER PR number + secret-alerts Secret-scanning alerts for one repo (one JSON line) + --owner OWNER GitHub owner (org or user) + --repo REPO Repository name + Token: HYPATIA_SECRET_ALERTS_TOKEN, else GITHUB_TOKEN. + Exit 3 when the alerts cannot be fetched. version Show version help Show this help diff --git a/lib/rules/secret_scanning_alerts.ex b/lib/rules/secret_scanning_alerts.ex index 88c2e226..7121ede7 100644 --- a/lib/rules/secret_scanning_alerts.ex +++ b/lib/rules/secret_scanning_alerts.ex @@ -14,8 +14,20 @@ defmodule Hypatia.Rules.SecretScanningAlerts do and the dismissal vocabulary (`revoked`, `used_in_tests`, `false_positive`) is the place to mark accepted ones. - Requires GITHUB_TOKEN with `secret_scanning_alerts: read` permission - (fine-grained PAT) or `security_events` scope (classic PAT). + ## Token + + A workflow's own `GITHUB_TOKEN` cannot read secret-scanning alerts: no + workflow permission covers them. They need a fine-grained PAT or GitHub + App token with `Secret scanning alerts: Read`. That token is read from + `HYPATIA_SECRET_ALERTS_TOKEN` first, then `GITHUB_TOKEN`, so a dedicated + read-only token never has to be exported under the general name. + + ## Errors + + The per-rule functions (`ssa001_open_alerts/2` and so on) log a fetch + failure and return `[]`, as the other API-backed rules do. `sweep/2` + instead returns `{:error, reason}`, so an estate-wide sweep can tell a + refused or disabled repo apart from a clean one. Rule IDs: SSA001-SSA004 """ @@ -44,32 +56,7 @@ defmodule Hypatia.Rules.SecretScanningAlerts do def ssa001_open_alerts(owner, repo) do case fetch_alerts(owner, repo) do {:ok, alerts} -> - alerts - |> Enum.filter(&(&1["state"] == "open")) - |> Enum.map(fn alert -> - secret_type = alert["secret_type_display_name"] || alert["secret_type"] || "unknown" - created = alert["created_at"] - age_days = age_in_days(created) - is_stale = age_days > @stale_threshold_days - - %{ - rule: "SSA001", - file: secret_type, - severity: :critical, - reason: build_alert_reason(secret_type, age_days, is_stale), - action: :escalate, - detail: %{ - alert_number: alert["number"], - secret_type: alert["secret_type"], - secret_type_display: alert["secret_type_display_name"], - age_days: age_days, - is_stale: is_stale, - created_at: created, - url: alert["html_url"], - locations_url: alert["locations_url"] - } - } - end) + open_alert_findings(alerts) {:error, reason} -> Logger.warning("SSA001: Failed to fetch secret-scanning alerts: #{reason}") @@ -77,6 +64,36 @@ defmodule Hypatia.Rules.SecretScanningAlerts do end end + @doc false + def open_alert_findings(alerts) do + alerts + |> Enum.filter(&(&1["state"] == "open")) + |> Enum.map(fn alert -> + secret_type = alert["secret_type_display_name"] || alert["secret_type"] || "unknown" + created = alert["created_at"] + age_days = age_in_days(created) + is_stale = age_days > @stale_threshold_days + + %{ + rule: "SSA001", + file: secret_type, + severity: :critical, + reason: build_alert_reason(secret_type, age_days, is_stale), + action: :escalate, + detail: %{ + alert_number: alert["number"], + secret_type: alert["secret_type"], + secret_type_display: alert["secret_type_display_name"], + age_days: age_days, + is_stale: is_stale, + created_at: created, + url: alert["html_url"], + locations_url: alert["locations_url"] + } + } + end) + end + # ─── SSA002: Severity summary ────────────────────────────────────────── @doc """ @@ -86,33 +103,35 @@ defmodule Hypatia.Rules.SecretScanningAlerts do """ def ssa002_severity_summary(owner, repo) do case fetch_alerts(owner, repo) do - {:ok, alerts} -> - open = Enum.filter(alerts, &(&1["state"] == "open")) - count = length(open) - - if count > 0 do - [ - %{ - rule: "SSA002", - file: "#{owner}/#{repo}", - severity: :critical, - reason: "#{count} open secret-scanning alert(s) -- rotate and purge from history", - action: :escalate, - detail: %{ - total: count, - by_type: - open - |> Enum.group_by(&(&1["secret_type"] || "unknown")) - |> Map.new(fn {k, v} -> {k, length(v)} end) - } - } - ] - else - [] - end + {:ok, alerts} -> summary_findings(owner, repo, alerts) + {:error, _} -> [] + end + end - {:error, _} -> - [] + @doc false + def summary_findings(owner, repo, alerts) do + open = Enum.filter(alerts, &(&1["state"] == "open")) + count = length(open) + + if count > 0 do + [ + %{ + rule: "SSA002", + file: "#{owner}/#{repo}", + severity: :critical, + reason: "#{count} open secret-scanning alert(s) -- rotate and purge from history", + action: :escalate, + detail: %{ + total: count, + by_type: + open + |> Enum.group_by(&(&1["secret_type"] || "unknown")) + |> Map.new(fn {k, v} -> {k, length(v)} end) + } + } + ] + else + [] end end @@ -126,38 +145,40 @@ defmodule Hypatia.Rules.SecretScanningAlerts do """ def ssa003_stale_alerts(owner, repo) do case fetch_alerts(owner, repo) do - {:ok, alerts} -> - alerts - |> Enum.filter(&(&1["state"] == "open")) - |> Enum.filter(fn alert -> - age_in_days(alert["created_at"]) > @stale_threshold_days - end) - |> Enum.map(fn alert -> - secret_type = alert["secret_type_display_name"] || alert["secret_type"] || "unknown" - age = age_in_days(alert["created_at"]) - - %{ - rule: "SSA003", - file: secret_type, - severity: :critical, - reason: - "Secret-scanning alert for #{secret_type} is #{age} days old " <> - "(threshold: #{@stale_threshold_days} days) -- overdue for rotation", - action: :escalate, - detail: %{ - alert_number: alert["number"], - secret_type: alert["secret_type"], - age_days: age, - threshold_days: @stale_threshold_days - } - } - end) - - {:error, _} -> - [] + {:ok, alerts} -> stale_findings(alerts) + {:error, _} -> [] end end + @doc false + def stale_findings(alerts) do + alerts + |> Enum.filter(&(&1["state"] == "open")) + |> Enum.filter(fn alert -> + age_in_days(alert["created_at"]) > @stale_threshold_days + end) + |> Enum.map(fn alert -> + secret_type = alert["secret_type_display_name"] || alert["secret_type"] || "unknown" + age = age_in_days(alert["created_at"]) + + %{ + rule: "SSA003", + file: secret_type, + severity: :critical, + reason: + "Secret-scanning alert for #{secret_type} is #{age} days old " <> + "(threshold: #{@stale_threshold_days} days) -- overdue for rotation", + action: :escalate, + detail: %{ + alert_number: alert["number"], + secret_type: alert["secret_type"], + age_days: age, + threshold_days: @stale_threshold_days + } + } + end) + end + # ─── SSA004: Dismissed without acceptable resolution ─────────────────── @doc """ @@ -168,39 +189,41 @@ defmodule Hypatia.Rules.SecretScanningAlerts do """ def ssa004_dismissed_without_fix(owner, repo) do case fetch_alerts(owner, repo) do - {:ok, alerts} -> - alerts - |> Enum.filter(fn a -> - a["state"] == "resolved" and - a["resolution"] not in @accepted_resolutions - end) - |> Enum.map(fn alert -> - secret_type = alert["secret_type_display_name"] || alert["secret_type"] || "unknown" - resolution = alert["resolution"] || "no reason given" - - %{ - rule: "SSA004", - file: secret_type, - severity: :high, - reason: - "Secret-scanning alert for #{secret_type} resolved as '#{resolution}' " <> - "-- confirm rotation completed and document acceptance reason", - action: :review, - detail: %{ - alert_number: alert["number"], - secret_type: alert["secret_type"], - resolution: resolution, - resolved_at: alert["resolved_at"], - resolution_comment: alert["resolution_comment"] - } - } - end) - - {:error, _} -> - [] + {:ok, alerts} -> dismissed_findings(alerts) + {:error, _} -> [] end end + @doc false + def dismissed_findings(alerts) do + alerts + |> Enum.filter(fn a -> + a["state"] == "resolved" and + a["resolution"] not in @accepted_resolutions + end) + |> Enum.map(fn alert -> + secret_type = alert["secret_type_display_name"] || alert["secret_type"] || "unknown" + resolution = alert["resolution"] || "no reason given" + + %{ + rule: "SSA004", + file: secret_type, + severity: :high, + reason: + "Secret-scanning alert for #{secret_type} resolved as '#{resolution}' " <> + "-- confirm rotation completed and document acceptance reason", + action: :review, + detail: %{ + alert_number: alert["number"], + secret_type: alert["secret_type"], + resolution: resolution, + resolved_at: alert["resolved_at"], + resolution_comment: alert["resolution_comment"] + } + } + end) + end + # ─── Comprehensive scan ──────────────────────────────────────────────── @doc """ @@ -208,10 +231,8 @@ defmodule Hypatia.Rules.SecretScanningAlerts do Returns `{:ok, result}` or `{:error, reason}`. """ def scan(owner, repo) do - token = System.get_env("GITHUB_TOKEN") - - if token == nil or token == "" do - {:error, "GITHUB_TOKEN not set -- cannot query secret-scanning alerts"} + if resolve_token() == nil do + {:error, missing_token_message()} else findings = ssa001_open_alerts(owner, repo) ++ @@ -219,21 +240,46 @@ defmodule Hypatia.Rules.SecretScanningAlerts do ssa003_stale_alerts(owner, repo) ++ ssa004_dismissed_without_fix(owner, repo) - deduped = - findings - |> Enum.uniq_by(fn f -> - {f.rule, Map.get(f.detail, :alert_number, f.file)} - end) - - {:ok, - %{ - findings: deduped, - total: length(deduped), - by_severity: group_by_severity(deduped) - }} + {:ok, result(findings)} end end + @doc """ + Run all secret-scanning checks for a repository with one API call, + returning `{:error, reason}` when the alerts cannot be fetched. + + Built for an estate-wide sweep: unlike `scan/2`, a refused token, a + repository with secret scanning disabled or a network failure is an + error, never an empty (clean-looking) result. + """ + def sweep(owner, repo) do + if resolve_token() == nil do + {:error, missing_token_message()} + else + case fetch_alerts(owner, repo) do + {:ok, alerts} -> {:ok, result(findings_from_alerts(owner, repo, alerts))} + {:error, reason} -> {:error, reason} + end + end + end + + @doc false + def findings_from_alerts(owner, repo, alerts) do + open_alert_findings(alerts) ++ + summary_findings(owner, repo, alerts) ++ + stale_findings(alerts) ++ + dismissed_findings(alerts) + end + + defp result(findings) do + deduped = + Enum.uniq_by(findings, fn f -> + {f.rule, Map.get(f.detail, :alert_number, f.file)} + end) + + %{findings: deduped, total: length(deduped), by_severity: group_by_severity(deduped)} + end + @doc """ Scan from a local repo path -- extracts owner/repo from git remote. """ @@ -246,41 +292,79 @@ defmodule Hypatia.Rules.SecretScanningAlerts do # ─── GitHub API ──────────────────────────────────────────────────────── + @doc false + # The dedicated token wins; the general one is the fallback. An empty + # value counts as unset, so `HYPATIA_SECRET_ALERTS_TOKEN=""` falls back. + def resolve_token do + Enum.find_value(["HYPATIA_SECRET_ALERTS_TOKEN", "GITHUB_TOKEN"], fn name -> + case System.get_env(name) do + token when is_binary(token) and token != "" -> token + _ -> nil + end + end) + end + + defp missing_token_message do + "GITHUB_TOKEN not set (nor HYPATIA_SECRET_ALERTS_TOKEN) -- " <> + "cannot query secret-scanning alerts" + end + defp fetch_alerts(owner, repo) do - token = System.get_env("GITHUB_TOKEN") + case resolve_token() do + nil -> + {:error, "GITHUB_TOKEN not set"} + + token -> + url = + "#{@github_api_base}/repos/#{owner}/#{repo}/secret-scanning/alerts" <> + "?per_page=#{@max_alerts_per_repo}" + + # No `-f`: on an HTTP error GitHub's JSON body carries the reason + # (for example "Secret scanning is disabled on this repository"), + # and `-f` would discard it. The status code is appended instead. + case System.cmd( + "curl", + [ + "-s", + "-w", + "\n%{http_code}", + "-H", + "Accept: application/vnd.github+json", + "-H", + "Authorization: Bearer #{token}", + "-H", + "X-GitHub-Api-Version: 2022-11-28", + url + ], + stderr_to_stdout: true + ) do + {output, 0} -> parse_response(output) + {error, _} -> {:error, "curl failed: #{String.slice(error, 0, 200)}"} + end + end + end - if token == nil or token == "" do - {:error, "GITHUB_TOKEN not set"} - else - url = - "#{@github_api_base}/repos/#{owner}/#{repo}/secret-scanning/alerts" <> - "?per_page=#{@max_alerts_per_repo}" - - case System.cmd( - "curl", - [ - "-s", - "-f", - "-H", - "Accept: application/vnd.github+json", - "-H", - "Authorization: Bearer #{token}", - "-H", - "X-GitHub-Api-Version: 2022-11-28", - url - ], - stderr_to_stdout: true - ) do - {body, 0} -> - case Jason.decode(body) do - {:ok, alerts} when is_list(alerts) -> {:ok, alerts} - {:ok, %{"message" => msg}} -> {:error, "GitHub API: #{msg}"} - {:error, _} -> {:error, "Invalid JSON response from GitHub API"} - end - - {error, _} -> - {:error, "curl failed: #{String.slice(error, 0, 200)}"} + @doc false + # `output` is the response body followed by a newline and the HTTP status. + def parse_response(output) do + {body, status} = + case String.split(output, "\n") |> Enum.reverse() do + [code | rest] -> {rest |> Enum.reverse() |> Enum.join("\n"), String.trim(code)} + [] -> {"", ""} end + + case {status, Jason.decode(body)} do + {"200", {:ok, alerts}} when is_list(alerts) -> + {:ok, alerts} + + {code, {:ok, %{"message" => msg}}} -> + {:error, "GitHub API #{code}: #{msg}"} + + {"200", _} -> + {:error, "Invalid JSON response from GitHub API"} + + {code, _} -> + {:error, "GitHub API returned HTTP #{code}"} end end diff --git a/test/cli_test.exs b/test/cli_test.exs index c95fca3d..ffa0d417 100644 --- a/test/cli_test.exs +++ b/test/cli_test.exs @@ -32,6 +32,33 @@ defmodule Hypatia.CLITest do end end + describe "secret-alerts" do + test "a successful fetch is one JSON line and exit 0, findings or not" do + sweep = fn "acme", "demo" -> + {:ok, %{findings: [], total: 0, by_severity: %{}}} + end + + assert {line, 0} = Hypatia.CLI.secret_alerts_output("acme", "demo", sweep) + assert {:ok, %{"status" => "ok", "repo" => "demo", "total" => 0}} = Jason.decode(line) + end + + test "a failed fetch is reported as an error with exit 3, never as clean" do + sweep = fn _, _ -> {:error, "GitHub API 404: Secret scanning is disabled"} end + + assert {line, 3} = Hypatia.CLI.secret_alerts_output("acme", "demo", sweep) + + assert {:ok, %{"status" => "error", "error" => "GitHub API 404: " <> _}} = + Jason.decode(line) + end + + test "help documents the command and its token" do + help = capture_io(fn -> Hypatia.CLI.main(["help"]) end) + + assert help =~ "secret-alerts" + assert help =~ "HYPATIA_SECRET_ALERTS_TOKEN" + end + end + describe "help output" do test "documents --exit-zero, HYPATIA_EXIT_ZERO, and exit codes" do help = capture_io(fn -> Hypatia.CLI.main(["help"]) end) diff --git a/test/secret_scanning_alerts_test.exs b/test/secret_scanning_alerts_test.exs index 8fd28105..572a0b6e 100644 --- a/test/secret_scanning_alerts_test.exs +++ b/test/secret_scanning_alerts_test.exs @@ -10,12 +10,17 @@ defmodule Hypatia.Rules.SecretScanningAlertsTest do alias Hypatia.Rules.SecretScanningAlerts + @token_vars ["GITHUB_TOKEN", "HYPATIA_SECRET_ALERTS_TOKEN"] + setup do - old_token = System.get_env("GITHUB_TOKEN") - System.delete_env("GITHUB_TOKEN") + saved = Map.new(@token_vars, &{&1, System.get_env(&1)}) + Enum.each(@token_vars, &System.delete_env/1) on_exit(fn -> - if old_token, do: System.put_env("GITHUB_TOKEN", old_token) + Enum.each(saved, fn + {name, nil} -> System.delete_env(name) + {name, value} -> System.put_env(name, value) + end) end) :ok @@ -57,6 +62,97 @@ defmodule Hypatia.Rules.SecretScanningAlertsTest do end end + describe "resolve_token/0" do + test "prefers the dedicated HYPATIA_SECRET_ALERTS_TOKEN" do + System.put_env("GITHUB_TOKEN", "test-general") + System.put_env("HYPATIA_SECRET_ALERTS_TOKEN", "test-dedicated") + assert SecretScanningAlerts.resolve_token() == "test-dedicated" + end + + test "falls back to GITHUB_TOKEN, treating an empty dedicated token as unset" do + System.put_env("GITHUB_TOKEN", "test-general") + System.put_env("HYPATIA_SECRET_ALERTS_TOKEN", "") + assert SecretScanningAlerts.resolve_token() == "test-general" + end + + test "is nil when neither is set" do + assert SecretScanningAlerts.resolve_token() == nil + end + end + + describe "sweep/2" do + test "is an error, not an empty result, when no token is set" do + assert {:error, msg} = SecretScanningAlerts.sweep("hyperpolymath", "test-nonexistent") + assert msg =~ "HYPATIA_SECRET_ALERTS_TOKEN" + end + end + + describe "parse_response/1" do + test "a 200 with a JSON list is the alert list" do + assert {:ok, [%{"number" => 1}]} = + SecretScanningAlerts.parse_response(~s([{"number":1}]) <> "\n200") + end + + test "an error status carries GitHub's message" do + body = ~s({"message":"Secret scanning is disabled on this repository."}) + + assert {:error, "GitHub API 404: Secret scanning is disabled on this repository."} = + SecretScanningAlerts.parse_response(body <> "\n404") + + assert {:error, "GitHub API 403: Resource not accessible by integration"} = + SecretScanningAlerts.parse_response( + ~s({"message":"Resource not accessible by integration"}) <> "\n403" + ) + end + + test "a non-JSON error body still reports the status" do + assert {:error, "GitHub API returned HTTP 502"} = + SecretScanningAlerts.parse_response("bad gateway\n502") + end + + test "a 200 that is not a list is an error, never an empty result" do + assert {:error, "Invalid JSON response from GitHub API"} = + SecretScanningAlerts.parse_response("not json\n200") + end + end + + describe "findings_from_alerts/3" do + defp days_ago(n), do: DateTime.utc_now() |> DateTime.add(-n * 86_400) |> DateTime.to_iso8601() + + test "builds SSA001-SSA004 from fetched alerts" do + alerts = [ + %{ + "number" => 1, + "state" => "open", + "secret_type" => "github_personal_access_token", + "created_at" => days_ago(30) + }, + %{ + "number" => 2, + "state" => "open", + "secret_type" => "aws_access_key_id", + "created_at" => days_ago(1) + }, + %{"number" => 3, "state" => "resolved", "resolution" => "wont_fix"}, + %{"number" => 4, "state" => "resolved", "resolution" => "revoked"} + ] + + findings = SecretScanningAlerts.findings_from_alerts("acme", "demo", alerts) + by_rule = Enum.group_by(findings, & &1.rule) + + assert length(by_rule["SSA001"]) == 2 + assert [%{file: "acme/demo", detail: %{total: 2}}] = by_rule["SSA002"] + # Only the 30-day-old alert is past the 7-day threshold. + assert [%{detail: %{alert_number: 1}}] = by_rule["SSA003"] + # `wont_fix` is not an accepted resolution; `revoked` is. + assert [%{detail: %{alert_number: 3}, severity: :high}] = by_rule["SSA004"] + end + + test "no alerts means no findings" do + assert SecretScanningAlerts.findings_from_alerts("acme", "demo", []) == [] + end + end + describe "scan_from_path/1" do test "returns error when remote is not a github URL" do tmp = Path.join(System.tmp_dir!(), "ssa-test-#{System.unique_integer([:positive])}")