mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
pulse-triage[bot]
2fb4bf9eee
Use a dedicated Auth Keys write client restricted to tag:ci-deploy, never the broader demo registration. Reject missing configuration and validate the actual running Self tag and DNS before Serve. Keep exact-source, denial-stop, no-replay and cleanup contracts, with connected offline registration and Serve controls. Change-source: pulse-maintainer
165 lines
8.6 KiB
Python
165 lines
8.6 KiB
Python
"""Offline controls for the actual browser workflow's narrow registration route."""
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
HELPER = ROOT / "scripts/release_browser_tailnet.py"
|
|
WORKFLOW = ROOT / ".github/workflows/qualify-browser-update-release.yml"
|
|
SPEC = importlib.util.spec_from_file_location("release_browser_tailnet", HELPER)
|
|
tailnet = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(tailnet)
|
|
|
|
|
|
def status():
|
|
return {"BackendState": "Running", "Self": {"Online": True,
|
|
"Tags": ["tag:ci-deploy"], "DNSName": "ci-browser.tawny-powan.ts.net."},
|
|
"Peer": {"private": {"DNSName": "private-peer", "private_data": "not-for-receipts"}}}
|
|
|
|
|
|
def workflow_step(name):
|
|
step = WORKFLOW.read_text().split(" - name: " + name + "\n", 1)[1].split(" - name:", 1)[0]
|
|
return textwrap.dedent(step.split(" run: |\n", 1)[1])
|
|
|
|
|
|
def assert_registration_binding(case, source):
|
|
# Closed, literal action inputs: no demo fallback or caller-selected tag/key.
|
|
action = source.split("uses: tailscale/github-action@", 1)[1].split(" - name:", 1)[0]
|
|
expected = {
|
|
"oauth-client-id": "${{ secrets.TS_CI_DEPLOY_OAUTH_CLIENT_ID }}",
|
|
"oauth-secret": "${{ secrets.TS_CI_DEPLOY_OAUTH_SECRET }}",
|
|
"tags": "tag:ci-deploy",
|
|
"version": "'1.94.2'",
|
|
}
|
|
actual = dict(line.strip().split(": ", 1) for line in action.split(" with:\n", 1)[1].splitlines() if line.strip())
|
|
case.assertEqual(expected, actual)
|
|
for broad in ("secrets.TS_OAUTH_CLIENT_ID", "secrets.TS_OAUTH_SECRET", "tags: tag:infra"):
|
|
case.assertNotIn(broad, source)
|
|
|
|
|
|
class SelfIdentityTest(unittest.TestCase):
|
|
def test_reads_only_the_running_narrow_self_identity(self):
|
|
self.assertEqual("https://ci-browser.tawny-powan.ts.net", tailnet.self_origin(status()))
|
|
observed = status()
|
|
observed["Peer"] = "untrusted and deliberately not a peer inventory"
|
|
self.assertEqual("https://ci-browser.tawny-powan.ts.net", tailnet.self_origin(observed))
|
|
observed["Self"]["DNSName"] = "a.tawny-powan.ts.net"
|
|
self.assertEqual("https://a.tawny-powan.ts.net", tailnet.self_origin(observed))
|
|
|
|
def test_rejects_broad_extra_missing_and_malformed_tags(self):
|
|
for tags in (["tag:infra"], ["tag:ci-deploy", "tag:infra"], [], None,
|
|
"tag:ci-deploy", ["tag:ci-deploy", "tag:ci-deploy"], ["tag:production"]):
|
|
with self.subTest(tags=tags), self.assertRaisesRegex(ValueError, "only tag:ci-deploy"):
|
|
observed = status()
|
|
observed["Self"]["Tags"] = tags
|
|
tailnet.self_origin(observed)
|
|
|
|
def test_rejects_non_self_wrong_tailnet_and_unsafe_dns(self):
|
|
for name in (None, 1, "private-peer", "ci.other.ts.net", "https://ci.tawny-powan.ts.net",
|
|
"ci.tawny-powan.ts.net\nsecret", "ci.tawny-powan.ts.net..",
|
|
"-ci.tawny-powan.ts.net", "ci-.tawny-powan.ts.net",
|
|
"a" * 64 + ".tawny-powan.ts.net"):
|
|
with self.subTest(name=name), self.assertRaises(ValueError):
|
|
observed = status()
|
|
observed["Self"]["DNSName"] = name
|
|
tailnet.self_origin(observed)
|
|
for observed in (None, [], {}, status() | {"Self": []}, status() | {"BackendState": "NeedsLogin"}):
|
|
with self.subTest(observed=observed), self.assertRaises(ValueError):
|
|
tailnet.self_origin(observed)
|
|
for online in (False, None, 1, "true"):
|
|
with self.subTest(online=online), self.assertRaises(ValueError):
|
|
observed = status()
|
|
observed["Self"]["Online"] = online
|
|
tailnet.self_origin(observed)
|
|
|
|
def test_cli_errors_never_return_raw_status_or_peers(self):
|
|
invalid = status()
|
|
invalid["Self"]["Tags"] = ["tag:infra"]
|
|
for raw in (json.dumps(invalid).encode(), b"private diagnostic is not JSON", b"\xff",
|
|
b" " * (tailnet.MAX_STATUS_BYTES + 1)):
|
|
with self.subTest(bytes=len(raw)):
|
|
result = subprocess.run(["python3", HELPER], input=raw, capture_output=True)
|
|
self.assertEqual(2, result.returncode)
|
|
self.assertEqual(b"", result.stdout)
|
|
self.assertEqual((tailnet.ERROR + "\n").encode(), result.stderr)
|
|
|
|
|
|
class WorkflowBindingTest(unittest.TestCase):
|
|
def test_only_dedicated_tag_scoped_oauth_binding_is_declared(self):
|
|
source = WORKFLOW.read_text()
|
|
assert_registration_binding(self, source)
|
|
self.assertIn("id: tailnet\n", source)
|
|
self.assertIn("if: always() && steps.tailnet.outcome == 'success'", source)
|
|
self.assertNotIn("steps.serve.outcome == 'success'", source)
|
|
|
|
def test_rejects_demo_credentials_and_broad_or_caller_selected_registration(self):
|
|
source = WORKFLOW.read_text()
|
|
for before, after in (
|
|
("secrets.TS_CI_DEPLOY_OAUTH_CLIENT_ID", "secrets.TS_OAUTH_CLIENT_ID"),
|
|
("secrets.TS_CI_DEPLOY_OAUTH_SECRET", "secrets.TS_OAUTH_SECRET"),
|
|
("tags: tag:ci-deploy", "tags: tag:infra"),
|
|
("tags: tag:ci-deploy", "tags: tag:ci-deploy,tag:infra"),
|
|
("tags: tag:ci-deploy", "tags: ${{ inputs.tags }}"),
|
|
("tags: tag:ci-deploy", "authkey: ${{ secrets.TS_OAUTH_SECRET }}"),
|
|
):
|
|
with self.subTest(after=after), self.assertRaises(AssertionError):
|
|
assert_registration_binding(self, source.replace(before, after))
|
|
|
|
def test_missing_narrow_binding_stops_even_when_demo_credentials_exist(self):
|
|
script = workflow_step("Require the dedicated narrow CI registration binding")
|
|
for client_id, secret in (("", ""), ("present", ""), ("", "present"), ("present", "present")):
|
|
env = {"PATH": os.environ["PATH"], "CI_CLIENT_ID": client_id, "CI_CLIENT_SECRET": secret,
|
|
"TS_OAUTH_CLIENT_ID": "broad-id-not-to-be-used", "TS_OAUTH_SECRET": "broad-secret-not-to-be-used"}
|
|
result = subprocess.run(["bash", "-c", script], env=env, capture_output=True, text=True)
|
|
self.assertEqual(0 if client_id and secret else 2, result.returncode)
|
|
self.assertEqual("", result.stdout)
|
|
self.assertNotIn("broad-id-not-to-be-used", result.stderr)
|
|
self.assertNotIn("broad-secret-not-to-be-used", result.stderr)
|
|
|
|
def test_actual_serve_step_stops_before_serve_and_output_on_wrong_identity(self):
|
|
script = workflow_step("Serve only this disposable install")
|
|
fixtures = [status()]
|
|
for tags in (["tag:infra"], ["tag:ci-deploy", "tag:infra"], []):
|
|
observed = status()
|
|
observed["Self"]["Tags"] = tags
|
|
fixtures.append(observed)
|
|
observed = status()
|
|
observed["Self"]["DNSName"] = "private-peer.other.ts.net"
|
|
fixtures.append(observed)
|
|
fixtures.append({})
|
|
for i, observed in enumerate(fixtures):
|
|
with self.subTest(fixture=i), tempfile.TemporaryDirectory() as tmp:
|
|
work = Path(tmp)
|
|
(work / "status.json").write_text(json.dumps(observed))
|
|
fake = work / "tailscale"
|
|
fake.write_text('#!/bin/bash\nset -eu\n'
|
|
'if [[ "$*" == "status --json" ]]; then cat "$FIXTURE/status.json"; '
|
|
'else printf "%s\\n" "$*" >> "$FIXTURE/serve-calls"; fi\n')
|
|
fake.chmod(0o700)
|
|
env = {"PATH": str(work) + ":" + os.environ["PATH"], "FIXTURE": tmp,
|
|
"GITHUB_OUTPUT": str(work / "output")}
|
|
result = subprocess.run(["bash", "-c", script], cwd=ROOT, env=env, capture_output=True, text=True)
|
|
self.assertEqual(0 if i == 0 else 2, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout)
|
|
self.assertNotIn("private-peer", result.stderr)
|
|
self.assertNotIn("not-for-receipts", result.stderr)
|
|
if i == 0:
|
|
self.assertEqual("serve --bg --https=443 http://127.0.0.1:7655\n", (work / "serve-calls").read_text())
|
|
header, value, delimiter = (work / "output").read_text().splitlines()
|
|
self.assertEqual("origin<<" + delimiter, header)
|
|
self.assertTrue(delimiter.startswith("pulse_output_"))
|
|
self.assertEqual("https://ci-browser.tawny-powan.ts.net", value)
|
|
else:
|
|
self.assertFalse((work / "serve-calls").exists())
|
|
self.assertFalse((work / "output").exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|