mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-26 05:29:45 +00:00
pulse-triage[bot]
8dcf561e4e
A normal-host enrolment report supplies lookup and host-report 401s, repeated installs and an absent inventory record. Document one named server-side registration check before touching the affected agent, with separate limits for persistence, credential hand-off and accepted reports. Contract-Neutral: Existing authentication help and its source-alignment controls only; no runtime, API, scope, route or credential-policy change. Change-source: pulse-maintainer
137 lines
7.7 KiB
Python
137 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Keep passive agent-authentication triage aligned with the real install flow.
|
|
|
|
These source/document controls do not authenticate an installed agent or
|
|
establish a reporter's cause. No credential, service or target is accessed.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
import re
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
DOC = ROOT / "docs/UNIFIED_AGENT.md"
|
|
|
|
|
|
def section():
|
|
text = DOC.read_text(encoding="utf-8")
|
|
marker = "### Agent lookup returns HTTP 401 or 403\n"
|
|
if marker not in text:
|
|
raise AssertionError("missing agent authentication triage")
|
|
return re.split(r"\n#{2,3} ", text.split(marker, 1)[1], 1)[0]
|
|
|
|
|
|
class AgentAuthDocsTest(unittest.TestCase):
|
|
def test_status_explanations_match_authentication_before_scoped_lookup(self):
|
|
text = " ".join(section().split())
|
|
routes = (ROOT / "internal/api/router_routes_registration.go").read_text()
|
|
self.assertIn(
|
|
'"/api/agents/agent/lookup", RequireAuth(r.config, '
|
|
'RequireScope(config.ScopeAgentReport, r.unifiedAgentHandlers.HandleLookup)',
|
|
routes,
|
|
)
|
|
auth = (ROOT / "internal/api/auth.go").read_text()
|
|
self.assertIn('http.Error(w, "Invalid API token", http.StatusUnauthorized)', auth)
|
|
scope = (ROOT / "internal/api/apihttp/scope.go").read_text()
|
|
denial = scope.split("func RespondMissingScope(", 1)[1]
|
|
self.assertIn("w.WriteHeader(http.StatusForbidden)", denial)
|
|
self.assertIn('"error":', denial)
|
|
self.assertIn('"missing_scope"', denial)
|
|
for fact in ("authenticates the request **before**", "**HTTP 401** is an authentication rejection",
|
|
"**HTTP 403** with `missing_scope`", "`agent:report`", "**HTTP 404**",
|
|
"upstream authentication layer"):
|
|
self.assertIn(fact, text)
|
|
|
|
def test_normal_install_is_not_conflated_with_bootstrap_enrolment(self):
|
|
text = " ".join(section().split())
|
|
install = (ROOT / "frontend-modern/src/components/Settings/useInfrastructureInstallState.tsx").read_text()
|
|
args = install.split("const getInstallerExtraArgs = () => [", 1)[1].split("];", 1)[0]
|
|
self.assertNotIn("--enroll", args)
|
|
scopes = (ROOT / "internal/api/agenttokens/install.go").read_text()
|
|
host = scopes.split("func HostScopes(", 1)[1].split("\n}", 1)[0]
|
|
self.assertIn("config.ScopeAgentReport", host)
|
|
self.assertIn("config.ScopeAgentConfigRead", host)
|
|
self.assertNotIn("ScopeAgentEnroll", host)
|
|
prompt = (ROOT / "frontend-modern/src/utils/agentInstallCommand.ts").read_text()
|
|
self.assertIn('read -r -s -p', prompt)
|
|
self.assertIn('--token-file "$token_file"', prompt)
|
|
for fact in ("silent token prompt", "does not change reporting permissions",
|
|
"It does not use `--enroll`", "bootstrap token for a runtime token",
|
|
"Do not substitute a bootstrap token", "normal host installation"):
|
|
self.assertIn(fact, text)
|
|
|
|
def test_triage_preserves_unknowns_state_and_private_credentials(self):
|
|
text = " ".join(section().split())
|
|
host = (ROOT / "internal/hostagent/agent.go").read_text()
|
|
self.assertIn("Pulse rejected this agent's API token", host)
|
|
client = (ROOT / "internal/remoteconfig/client.go").read_text()
|
|
self.assertIn('Str("component", "remote_config_client")', client)
|
|
for fact in ("bounded agent log reader", "affected agent host", "retrieval and does not establish",
|
|
"ordinary host reports were accepted", "successful-looking installer output does not prove",
|
|
"No host-report error", "inconclusive", "Do not restart, re-enrol, delete identity",
|
|
"same Pulse instance", "do not prove that the running service is using that record",
|
|
"service environments, token files and connection files private",
|
|
"Do not grant administrator scope", "disable authentication/TLS",
|
|
"manually redacted error line", "saved credentials unchanged",
|
|
"not itself a reason to reinstall or rotate every token"):
|
|
self.assertIn(fact, text)
|
|
self.assertNotIn("```", section(), "Reuse the bounded reader; do not add live credential probes")
|
|
|
|
def test_shipped_help_matches_the_canonical_guide(self):
|
|
self.assertEqual(DOC.read_bytes(),
|
|
(ROOT / "frontend-modern/public/docs/UNIFIED_AGENT.md").read_bytes())
|
|
|
|
def test_inventory_guidance_matches_available_navigation_and_metadata(self):
|
|
text = " ".join(section().split())
|
|
catalog = (ROOT / "frontend-modern/src/components/Settings/settingsNavCatalog.ts").read_text()
|
|
api_entry = re.search(r"id: 'api',\s+label: '([^']+)'", catalog)
|
|
self.assertIsNotNone(api_entry)
|
|
self.assertIn(f"Settings → {api_entry.group(1)}", text)
|
|
inventory = (ROOT / "frontend-modern/src/components/Settings/APITokenManager.tsx").read_text()
|
|
for field in ("Created", "Last used", "Scopes"):
|
|
self.assertIn(field, inventory)
|
|
self.assertIn(f"**{field}**", text)
|
|
self.assertIn("Security group", text)
|
|
self.assertNotIn("Security → API Tokens", text)
|
|
self.assertNotIn("presence, expiry", text)
|
|
self.assertIn("not a token-use audit", text)
|
|
self.assertIn("do not share the token or its hint", text)
|
|
|
|
def test_dual_rejection_does_not_recommend_another_install_or_assume_a_restore(self):
|
|
text = " ".join(section().split())
|
|
for fact in ("both lookup and host reporting return 401", "not a lookup-only failure",
|
|
"restore/upgrade explanation is generic", "not an established cause",
|
|
"already tried reinstalling", "do not repeat it",
|
|
"missing record", "service is presenting the same token"):
|
|
self.assertIn(fact, text)
|
|
|
|
def test_missing_record_check_registers_once_before_touching_the_agent(self):
|
|
text = " ".join(section().split())
|
|
install = (ROOT / "frontend-modern/src/components/Settings/useInfrastructureInstallState.tsx").read_text()
|
|
generation = install.split("const generateInstallToken = async (", 1)[1].split(
|
|
"const handleGenerateToken = async", 1)[0]
|
|
self.assertIn("NodesAPI.createHostAgentInstallToken", generation)
|
|
self.assertIn("enableCommands: withCommands", generation)
|
|
self.assertNotIn("deleteToken", generation)
|
|
ui = (ROOT / "frontend-modern/src/components/Settings/InfrastructureInstallerSection.tsx").read_text()
|
|
self.assertIn("Generate token", ui)
|
|
inventory = (ROOT / "internal/api/security_tokens.go").read_text()
|
|
listing = inventory.split("func (r *Router) handleListAPITokens(", 1)[1].split(
|
|
"// handleGetAPIToken", 1)[0]
|
|
self.assertIn("range r.config.APITokens", listing)
|
|
self.assertIn("toAPITokenDTO(record)", listing)
|
|
for fact in ("If the intended record is absent", "server registration",
|
|
"one new install token a distinct name", "Before running any install command",
|
|
"refresh the page", "does not change the Pi or any other agent's saved credential",
|
|
"Leave command execution disabled", "missing or disappears, stop there",
|
|
"do not generate more tokens", "generation showed an error",
|
|
"not persistence across a server restart", "successful credential hand-off",
|
|
"accepted host reports", "Retain the existing installation",
|
|
"not, by itself, a reason to mint a replacement"):
|
|
self.assertIn(fact, text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|