Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 118 additions & 0 deletions .github/scripts/release_guard.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
"""Verify immutable chart source and published, attested operator dependencies."""

import argparse
import base64
import json
import os
from pathlib import Path
import re
import subprocess


VERSION = r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)(?:-[0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*)?"


def require(condition, message):
if not condition:
raise ValueError(message)


def run(*args):
return subprocess.check_output(args, text=True).strip()


def api(path):
return json.loads(run("gh", "api", path))


def scalar(chart, key):
values = re.findall(r"^" + key + r":\s*['\"]?([^'\"\s]+)['\"]?\s*$", chart, re.M)
require(len(values) == 1, f"expected one normalized {key}")
require(re.fullmatch(VERSION, values[0]), f"invalid {key}")
return values[0]


def published_release(repository, tag):
release = api(f"repos/{repository}/releases/tags/{tag}")
require(release.get("tag_name") == tag and release.get("draft") is False
and release.get("published_at"), f"{repository} {tag} is not published")
return release


def tag_commit(repository, tag):
obj = api(f"repos/{repository}/git/ref/tags/{tag}")["object"]
for _ in range(5):
if obj["type"] == "commit":
require(re.fullmatch(r"[a-f0-9]{40}", obj["sha"]), "invalid source SHA")
return obj["sha"]
require(obj["type"] == "tag", "release tag does not resolve to a commit")
obj = api(f"repos/{repository}/git/tags/{obj['sha']}")["object"]
raise ValueError("release tag nesting limit exceeded")


def image_digest(image):
digest = run("oras", "resolve", image)
require(re.fullmatch(r"sha256:[a-f0-9]{64}", digest), "invalid image digest")
return digest


def verify_image_pin(chart, digest):
metadata = run("helm", "show", "chart", chart)
pins = re.findall(r'^ cfgate\.io/operator-image-digest:\s*[\'"]?(sha256:[a-f0-9]{64})[\'"]?\s*$', metadata, re.M)
require(pins == [digest], "chart operator pin does not match verified release digest")
deployment = run("helm", "template", "cfgate", chart, "--show-only", "templates/deployment.yaml")
images = re.findall(r'^\s+image:\s*[\'"]?([^\s\'"]+)[\'"]?\s*$', deployment, re.M)
require(images == ["ghcr.io/cfgate/cfgate@" + digest],
"default Deployment must install the verified operator digest")


def verify(tag, sha, package=None):
require(re.fullmatch("v" + VERSION, tag), "invalid chart release tag")
require(re.fullmatch(r"[a-f0-9]{40}", sha), "invalid chart source SHA")
require(run("git", "rev-parse", "HEAD") == sha, "checkout does not match event SHA")
require(run("git", "rev-parse", f"refs/tags/{tag}^{{commit}}") == sha,
"chart tag does not match event SHA")
remote = run("git", "ls-remote", "--tags", "origin", f"refs/tags/{tag}", f"refs/tags/{tag}^{{}}")
refs = dict(line.split()[::-1] for line in remote.splitlines())
require(refs.get(f"refs/tags/{tag}^{{}}", refs.get(f"refs/tags/{tag}")) == sha,
"remote chart tag moved or disappeared")
metadata = run("helm", "show", "chart", ".")
version, app = scalar(metadata, "version"), scalar(metadata, "appVersion")
require(tag == "v" + version, "Chart.yaml version does not match tag")
published_release("cfgate/cfgate", "v" + app)
operator_sha = tag_commit("cfgate/cfgate", "v" + app)
image = "ghcr.io/cfgate/cfgate:" + app
digest = image_digest(image)
Comment thread
greptile-apps[bot] marked this conversation as resolved.
Resolved
verify_image_pin(".", digest)
if package is not None:
verify_image_pin(package, digest)
# These checks bind the available OCI digest to the released source, rather
# than trusting the mutable image tag or the release's target_commitish.
run("gh", "attestation", "verify", "oci://ghcr.io/cfgate/cfgate@" + digest,
"--bundle-from-oci",
"--repo", "cfgate/cfgate", "--signer-workflow", "cfgate/cfgate/.github/workflows/release.yml",
"--source-ref", "refs/tags/v" + app, "--source-digest", operator_sha)
source = api(f"repos/cfgate/cfgate/contents/internal/cloudflared/deployment.go?ref={operator_sha}")
require(source.get("encoding") == "base64", "unexpected source encoding")
code = base64.b64decode(source["content"]).decode()
matches = re.findall(r'DefaultImage\s*=\s*"(ghcr\.io/inherent-design/cloudflared:([0-9A-Za-z.-]+))"', code)
require(len(matches) == 1, "released operator must identify one versioned connector")
fork_image, fork_version = matches[0]
require(re.fullmatch(VERSION, fork_version), "invalid connector version")
published_release("inherent-design/cloudflared", "v" + fork_version)
fork_digest = image_digest(fork_image)
return {"version": version, "source": sha, "app_version": app,
"operator_source": operator_sha, "operator_image": image, "operator_digest": digest,
"connector_image": fork_image, "connector_digest": fork_digest}


if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--package", help="also verify the packaged chart's default image")
args = parser.parse_args()
evidence = verify(os.environ["RELEASE_TAG"], os.environ["RELEASE_SHA"], args.package)
Path("dist").mkdir(exist_ok=True)
Path("dist/release-dependencies.json").write_text(json.dumps(evidence, indent=2) + "\n")
with open(os.environ["GITHUB_OUTPUT"], "a") as output:
output.write("clean=" + evidence["version"] + "\n")
print(json.dumps(evidence, indent=2))
143 changes: 143 additions & 0 deletions .github/scripts/test_chart.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,143 @@
"""Exercise real Helm rendering and schema rejection without a cluster."""

import json
from pathlib import Path
import re
import shutil
import subprocess
import tempfile
import unittest


ROOT = Path(__file__).resolve().parents[2]
DIGEST = "sha256:3d3eaeae0ae0a76f8b3bc5271f642cf06f42e6525c85cb16e1b778fb4d864d6a"


class ChartTests(unittest.TestCase):
def render(self, values=None, chart=".", deployment=True):
with tempfile.NamedTemporaryFile(mode="w", suffix=".json") as f:
json.dump(values or {}, f)
f.flush()
args = ["helm", "template", "cfgate", chart, "--namespace", "manager", "-f", f.name]
if deployment:
args += ["--show-only", "templates/deployment.yaml"]
return subprocess.run(args, cwd=ROOT, capture_output=True, text=True)

def image(self, values=None, chart="."):
result = self.render(values, chart)
self.assertEqual(result.returncode, 0, result.stderr)
images = re.findall(r'^\s+image: "([^"]+)"$', result.stdout, re.M)
self.assertEqual(len(images), 1)
return images[0]

def test_image_selection(self):
custom = "sha256:" + "a" * 64
for values, expected in [
({}, "ghcr.io/cfgate/cfgate@" + DIGEST),
({"image": {"tag": "custom"}}, "ghcr.io/cfgate/cfgate:custom"),
({"image": {"repository": "example.com/operator"}}, "example.com/operator:0.2.0-alpha.6"),
({"image": {"repository": "example.com/operator", "tag": "custom"}}, "example.com/operator:custom"),
({"image": {"digest": custom, "tag": "ignored"}}, "ghcr.io/cfgate/cfgate@" + custom),
({"image": {"repository": "example.com/operator", "digest": custom}}, "example.com/operator@" + custom),
]:
with self.subTest(values=values):
self.assertEqual(self.image(values), expected)

def test_controller_namespace_and_service_account_overrides(self):
result = self.render({
"installCRDs": False,
"namespaceOverride": "workloads",
"controller": {"clusterDomain": "cluster.internal", "installationNamespace": "claims",
"cloudflareRequestTimeoutSeconds": 20, "maxIngressRules": 2000,
"maxConfigurationBytes": 2097152},
"serviceAccount": {"create": False, "name": "external-manager"},
"metrics": {"port": 9090}, "health": {"port": 9091},
}, deployment=False)
self.assertEqual(result.returncode, 0, result.stderr)
for expected in ['namespace: workloads', 'serviceAccountName: external-manager',
'name: external-manager', 'fieldPath: metadata.namespace', 'enableServiceLinks: false',
'--cluster-domain=cluster.internal', '--installation-namespace=claims',
'--cloudflare-request-timeout=20s', '--max-ingress-rules=2000',
'--max-configuration-bytes=2097152', '--metrics-bind-address=:9090',
'--health-probe-bind-address=:9091']:
self.assertIn(expected, result.stdout)
self.assertNotIn('ServiceAccount', re.findall(r'^kind: (\S+)$', result.stdout, re.M))
binding = next(doc for doc in result.stdout.split('---') if '\nkind: ClusterRoleBinding\n' in doc)
self.assertIn('name: external-manager\n namespace: workloads', binding)

def test_old_values_and_packaged_defaults(self):
# Old releases lack controller and image.digest values. Reuse must still
# select this chart's own pin, not retain the old chart's release digest.
result = self.render({"controller": None, "image": {"tag": "", "digest": None}})
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('ghcr.io/cfgate/cfgate@' + DIGEST, result.stdout)
self.assertIn('--cloudflare-request-timeout=30s', result.stdout)
self.assertNotIn('--installation-namespace=', result.stdout)
with tempfile.TemporaryDirectory() as directory:
subprocess.run(["helm", "package", ".", "-d", directory], cwd=ROOT, check=True, capture_output=True)
package, = Path(directory).glob('*.tgz')
self.assertEqual(self.image(chart=str(package)), 'ghcr.io/cfgate/cfgate@' + DIGEST)

def test_chart_pin_changes_and_missing_pin_rejects(self):
with tempfile.TemporaryDirectory() as directory:
chart = Path(directory)
for name in ['Chart.yaml', 'values.yaml', 'values.schema.json']:
shutil.copyfile(ROOT / name, chart / name)
shutil.copytree(ROOT / 'templates', chart / 'templates')
metadata = (chart / 'Chart.yaml').read_text()
updated = 'sha256:' + 'b' * 64
(chart / 'Chart.yaml').write_text(metadata.replace(DIGEST, updated))
self.assertEqual(self.image({"image": {"tag": "", "digest": None}}, directory),
'ghcr.io/cfgate/cfgate@' + updated)
for invalid in ['', 'sha256:bad']:
(chart / 'Chart.yaml').write_text(metadata.replace(DIGEST, invalid))
result = self.render(chart=directory)
self.assertNotEqual(result.returncode, 0)
self.assertIn('digest', result.stderr)

def test_external_crds_and_rbac(self):
result = self.render({"installCRDs": False, "rbac": {"create": False}}, deployment=False)
self.assertEqual(result.returncode, 0, result.stderr)
for kind in ['CustomResourceDefinition', 'ClusterRole', 'ClusterRoleBinding', 'Role', 'RoleBinding']:
self.assertNotIn('kind: ' + kind + '\n', result.stdout)

def test_claim_permissions_follow_installation_namespace(self):
for values, namespace, account_namespace, account in [
({}, 'manager', 'manager', 'cfgate'),
({'namespaceOverride': 'workloads'}, 'workloads', 'workloads', 'cfgate'),
({'controller': None}, 'manager', 'manager', 'cfgate'),
({'namespaceOverride': 'workloads', 'controller': {'installationNamespace': 'claims'},
'serviceAccount': {'create': False, 'name': 'external-manager'}},
'claims', 'workloads', 'external-manager'),
]:
with self.subTest(values=values):
result = self.render(values, deployment=False)
self.assertEqual(result.returncode, 0, result.stderr)
docs = {re.search(r'^kind: (\S+)$', doc, re.M)[1]: doc
for doc in result.stdout.split('---') if re.search(r'^kind: (\S+)$', doc, re.M)}
self.assertNotIn('configmaps', docs['ClusterRole'])
role, binding = docs['Role'], docs['RoleBinding']
self.assertIn(' namespace: ' + namespace + '\n', role)
self.assertIn(' namespace: ' + namespace + '\n', binding)
self.assertIn('resources: ["configmaps"]', role)
self.assertIn('verbs: ["get", "create", "delete"]', role)
self.assertIn(' kind: Role\n name: cfgate-claims', binding)
self.assertIn('name: ' + account + '\n namespace: ' + account_namespace, binding)

def test_invalid_values_rejected(self):
for values in [
{"image": {"digest": "latest"}}, {"image": {"digest": "sha256:abc"}},
{"controller": {"cloudflareRequestTimeoutSeconds": 0}},
{"controller": {"cloudflareRequestTimeoutSeconds": 9223372037}},
{"controller": {"maxIngressRules": 0}}, {"controller": {"maxConfigurationBytes": 0}},
{"controller": {"clusterDomain": "bad_domain"}},
{"controller": {"installationNamespace": "Bad"}},
]:
with self.subTest(values=values):
result = self.render(values)
self.assertNotEqual(result.returncode, 0)
self.assertIn('schema', result.stderr.lower())


if __name__ == "__main__":
unittest.main()
Loading