Integrate reviewed PBS safety fixture repairs

Merge exact candidate 0955ddf9e0. Keep canonical API-only least-privilege, failure-stop and unsafe token/pipe assertions alongside the broader shell-fence checks and explicit TLS fixture floor.

Contract-Neutral: Test-only conflict resolution preserves existing PBS API and agent installation contracts.
Change-source: pulse-maintainer
This commit is contained in:
pulse-triage[bot] 2026-10-01 10:31:47 +01:00
commit ac34b6f461
2 changed files with 58 additions and 16 deletions

View file

@ -1,11 +1,32 @@
package repoctl
import (
"regexp"
"strings"
"testing"
)
func pbsShellExamples(document string) string {
var examples []string
blocks := strings.Split(document, "```")
for i := 1; i < len(blocks); i += 2 {
language, body, _ := strings.Cut(blocks[i], "\n")
switch strings.TrimSpace(language) {
case "bash", "sh", "shell":
examples = append(examples, body)
}
}
return strings.Join(examples, "\n")
}
func TestPBSShellExamplesKeepWarningsSeparateFromCommands(t *testing.T) {
document := "Do not use `--insecure` or `-k`.\n```bash\ncurl --fail https://pbs.example.com\n```\n" +
"```sh\ncurl --insecure https://pbs.example.com\n```\n"
examples := pbsShellExamples(document)
if examples != "curl --fail https://pbs.example.com\n\ncurl --insecure https://pbs.example.com\n" {
t.Fatalf("unexpected shell examples: %q", examples)
}
}
func TestSetupBootstrapDocsStayOnCanonicalArtifactContract(t *testing.T) {
apiRel := "docs/API.md"
apiDoc := readRepoFile(t, apiRel)
@ -24,21 +45,33 @@ func TestSetupBootstrapDocsStayOnCanonicalArtifactContract(t *testing.T) {
"### Method 1: API-Only Connection (Recommended)",
"proxmox-backup-manager acl update / Audit --auth-id pulse-monitor@pbs",
"proxmox-backup-manager acl update / Audit --auth-id 'pulse-monitor@pbs!pulse-token'",
`chmod 600 "$HOME/.config/pulse/pbs-agent-token"`,
`--output "$HOME/.config/pulse/pbs-agent-install.sh"`,
"https://pulse.example.com/install.sh",
`--token-file "$HOME/.config/pulse/pbs-agent-token"`,
"Stop if the download fails",
"Create a separate Pulse token in **API Access**",
"enter an administrator root shell",
`chmod 700 "$HOME/.config/pulse"`,
`chmod 600 "$HOME/.config/pulse/pbs-agent-token"`,
`vi "$HOME/.config/pulse/pbs-agent-token"`,
`curl --fail --silent --show-error`,
`--output "$HOME/.config/pulse/pbs-agent-install.sh"`,
`https://pulse.example.com/install.sh`,
"successful download and inspection",
`bash "$HOME/.config/pulse/pbs-agent-install.sh"`,
`--token-file "$HOME/.config/pulse/pbs-agent-token"`,
`--enable-proxmox --proxmox-type pbs --enable-docker=false`,
"Do not bypass certificate checks",
})
// Inspect executable fences, not prose that warns against unsafe flags.
commands := strings.Join(regexp.MustCompile("(?s)```bash\n(.*?)```").FindAllString(pbsDoc, -1), "\n")
assertContainsNone(t, pbsRel, commands, []string{
"PULSE_SETUP_TOKEN=",
"sudo env PULSE_SETUP_TOKEN=",
"curl -k",
"--insecure",
shellExamples := pbsShellExamples(pbsDoc)
if shellExamples == "" {
t.Fatal("PBS guide must retain executable shell examples")
}
assertContainsNone(t, pbsRel, shellExamples, []string{
`curl -sSL "http://<pulse-ip>:7655/api/setup-script?type=pbs&host=https://<pbs-ip>:8007&pulse_url=http://<pulse-ip>:7655" | bash`,
`PULSE_SETUP_TOKEN=`,
`sudo env PULSE_SETUP_TOKEN=`,
`--insecure`,
` -k`,
`curl -k`,
`--token "`,
"| bash",
`| bash`,
})
}

View file

@ -62,6 +62,13 @@ def certificate(directory, name):
return certificate, key
def server_context(cert, key):
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.load_cert_chain(str(cert), str(key))
return context
@contextmanager
def server(cert, key, status=200, installer=None, installer_path="/install.sh"):
requests = []
@ -78,9 +85,7 @@ def server(cert, key, status=200, installer=None, installer_path="/install.sh"):
pass
http = HTTPServer(("127.0.0.1", 0), Handler)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.load_cert_chain(str(cert), str(key))
context = server_context(cert, key)
http.socket = context.wrap_socket(http.socket, server_side=True)
thread = threading.Thread(target=http.serve_forever, daemon=True)
thread.start()
@ -103,6 +108,10 @@ class PBSDocsTest(unittest.TestCase):
def tearDownClass(cls):
cls.temporary.cleanup()
def test_tls_fixture_explicitly_rejects_legacy_protocols(self):
self.assertEqual(server_context(self.cert, self.key).minimum_version,
ssl.TLSVersion.TLSv1_2)
def test_shipped_copy_and_safe_scope(self):
self.assertEqual(DOC.read_bytes(), (ROOT / "frontend-modern/public/docs/PBS.md").read_bytes())
text = DOC.read_text()