feat(guanghu-os): stage independent Forgejo continuity
This commit is contained in:
parent
ace87cfba7
commit
e4750c57a7
10 changed files with 467 additions and 3 deletions
|
|
@ -22,6 +22,7 @@ grep -Fq -- '--fail-under-lines 100' "${runner}"
|
|||
grep -Fq -- '--fail-under-functions 100' "${runner}"
|
||||
grep -Fq -- '--test broadcast_library' "${runner}"
|
||||
grep -Fq 'test-native-public-projection.sh' "${runner}"
|
||||
grep -Fq 'test-independent-forgejo-shadow-verifier.py' "${runner}"
|
||||
grep -Fq 'GHNQG_PASS_100' "${runner}"
|
||||
grep -Fq 'GHNQG_FAIL_0' "${runner}"
|
||||
grep -Fq 'total_score: ${total_score}' "${runner}"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,125 @@
|
|||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import pathlib
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
||||
SCRIPT = pathlib.Path(__file__).with_name("verify-independent-forgejo-shadow.py")
|
||||
SPEC = importlib.util.spec_from_file_location("forgejo_shadow_verifier", SCRIPT)
|
||||
assert SPEC is not None and SPEC.loader is not None
|
||||
MODULE = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(MODULE)
|
||||
|
||||
|
||||
class IndependentForgejoShadowVerifierTest(unittest.TestCase):
|
||||
def test_exact_shadow_passes_and_head_drift_fails(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
root = pathlib.Path(temporary)
|
||||
binary = root / "forgejo"
|
||||
binary.write_bytes(b"fixed-forgejo-binary")
|
||||
repository = root / "repositories" / "bingshuo" / "example.git"
|
||||
repository.mkdir(parents=True)
|
||||
subprocess.run(["git", "init", "--bare", "-q", str(repository)], check=True)
|
||||
tree = subprocess.run(
|
||||
["git", f"--git-dir={repository}", "mktree"],
|
||||
input="",
|
||||
capture_output=True,
|
||||
check=True,
|
||||
text=True,
|
||||
).stdout.strip()
|
||||
commit = subprocess.run(
|
||||
["git", f"--git-dir={repository}", "commit-tree", tree],
|
||||
input="source\n",
|
||||
capture_output=True,
|
||||
check=True,
|
||||
text=True,
|
||||
env={
|
||||
"GIT_AUTHOR_NAME": "test",
|
||||
"GIT_AUTHOR_EMAIL": "test@example.invalid",
|
||||
"GIT_AUTHOR_DATE": "2001-01-01T00:00:00+00:00",
|
||||
"GIT_COMMITTER_NAME": "test",
|
||||
"GIT_COMMITTER_EMAIL": "test@example.invalid",
|
||||
"GIT_COMMITTER_DATE": "2001-01-01T00:00:00+00:00",
|
||||
},
|
||||
).stdout.strip()
|
||||
subprocess.run(
|
||||
["git", f"--git-dir={repository}", "update-ref", "refs/heads/main", commit],
|
||||
check=True,
|
||||
)
|
||||
database = root / "forgejo.db"
|
||||
connection = sqlite3.connect(database)
|
||||
connection.executescript(
|
||||
"CREATE TABLE repository(id INTEGER); INSERT INTO repository VALUES(1);"
|
||||
"CREATE TABLE user(id INTEGER); INSERT INTO user VALUES(1);"
|
||||
)
|
||||
connection.close()
|
||||
manifest = {
|
||||
"schema": MODULE.SCHEMA,
|
||||
"source_node": "JD-FD-PRIMARY",
|
||||
"forgejo_version": "16.0.1",
|
||||
"binary_sha256": hashlib.sha256(binary.read_bytes()).hexdigest(),
|
||||
"database_repository_count": 1,
|
||||
"database_user_count": 1,
|
||||
"repositories": [{"path": "bingshuo/example.git", "head": commit}],
|
||||
}
|
||||
manifest_path = root / "manifest.json"
|
||||
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
||||
loaded = MODULE.load_manifest(manifest_path)
|
||||
self.assertEqual(loaded["source_node"], "JD-FD-PRIMARY")
|
||||
self.assertEqual(
|
||||
MODULE.repository_heads(root / "repositories"),
|
||||
{"bingshuo/example.git": commit},
|
||||
)
|
||||
self.assertEqual(MODULE.database_readback(database), ("ok", 1, 1))
|
||||
drift = subprocess.run(
|
||||
["git", f"--git-dir={repository}", "commit-tree", tree, "-p", commit],
|
||||
input="drift\n",
|
||||
capture_output=True,
|
||||
check=True,
|
||||
text=True,
|
||||
env={
|
||||
"GIT_AUTHOR_NAME": "test",
|
||||
"GIT_AUTHOR_EMAIL": "test@example.invalid",
|
||||
"GIT_AUTHOR_DATE": "2001-01-02T00:00:00+00:00",
|
||||
"GIT_COMMITTER_NAME": "test",
|
||||
"GIT_COMMITTER_EMAIL": "test@example.invalid",
|
||||
"GIT_COMMITTER_DATE": "2001-01-02T00:00:00+00:00",
|
||||
},
|
||||
).stdout.strip()
|
||||
subprocess.run(
|
||||
["git", f"--git-dir={repository}", "update-ref", "refs/heads/main", drift],
|
||||
check=True,
|
||||
)
|
||||
self.assertNotEqual(
|
||||
MODULE.repository_heads(root / "repositories"),
|
||||
{"bingshuo/example.git": commit},
|
||||
)
|
||||
|
||||
def test_manifest_rejects_escape_and_duplicates(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
path = pathlib.Path(temporary) / "manifest.json"
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema": MODULE.SCHEMA,
|
||||
"repositories": [
|
||||
{"path": "../escape.git", "head": "1" * 40},
|
||||
{"path": "../escape.git", "head": "1" * 40},
|
||||
],
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
with self.assertRaises(ValueError):
|
||||
MODULE.load_manifest(path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -0,0 +1,145 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Verify one independent Forgejo shadow against a source-side manifest."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import pathlib
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import urllib.request
|
||||
from typing import Any
|
||||
|
||||
|
||||
SCHEMA = "guanghu.independent-forgejo-shadow-manifest/v1"
|
||||
|
||||
|
||||
def sha256(path: pathlib.Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as source:
|
||||
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def load_manifest(path: pathlib.Path) -> dict[str, Any]:
|
||||
value = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(value, dict) or value.get("schema") != SCHEMA:
|
||||
raise ValueError("source manifest schema mismatch")
|
||||
repositories = value.get("repositories")
|
||||
if not isinstance(repositories, list) or not repositories:
|
||||
raise ValueError("source manifest must contain repositories")
|
||||
paths: set[str] = set()
|
||||
for repository in repositories:
|
||||
if not isinstance(repository, dict):
|
||||
raise ValueError("repository manifest entry must be an object")
|
||||
path_value = repository.get("path")
|
||||
head = repository.get("head")
|
||||
if (
|
||||
not isinstance(path_value, str)
|
||||
or path_value.startswith("/")
|
||||
or ".." in pathlib.PurePosixPath(path_value).parts
|
||||
or not path_value.endswith(".git")
|
||||
or path_value in paths
|
||||
):
|
||||
raise ValueError("repository path is unsafe or duplicated")
|
||||
if not isinstance(head, str) or len(head) != 40:
|
||||
raise ValueError("repository head must be a full commit")
|
||||
paths.add(path_value)
|
||||
return value
|
||||
|
||||
|
||||
def repository_heads(root: pathlib.Path) -> dict[str, str]:
|
||||
result: dict[str, str] = {}
|
||||
for repository in sorted(root.glob("*/*.git")):
|
||||
if repository.is_symlink() or not repository.is_dir():
|
||||
raise ValueError("repository must be a real directory")
|
||||
completed = subprocess.run(
|
||||
["git", f"--git-dir={repository}", "rev-parse", "refs/heads/main"],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=10,
|
||||
)
|
||||
result[repository.relative_to(root).as_posix()] = completed.stdout.strip()
|
||||
return result
|
||||
|
||||
|
||||
def database_readback(path: pathlib.Path) -> tuple[str, int, int]:
|
||||
connection = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
|
||||
try:
|
||||
integrity = str(connection.execute("PRAGMA integrity_check").fetchone()[0])
|
||||
repositories = int(connection.execute("SELECT count(*) FROM repository").fetchone()[0])
|
||||
users = int(connection.execute("SELECT count(*) FROM user").fetchone()[0])
|
||||
return integrity, repositories, users
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
|
||||
def http_version(url: str) -> str:
|
||||
with urllib.request.urlopen(url, timeout=5) as response: # noqa: S310
|
||||
if response.status != 200:
|
||||
raise ValueError("Forgejo version endpoint did not return 200")
|
||||
value = json.load(response)
|
||||
if not isinstance(value, dict) or not isinstance(value.get("version"), str):
|
||||
raise ValueError("Forgejo version response is malformed")
|
||||
return value["version"]
|
||||
|
||||
|
||||
def verify(args: argparse.Namespace) -> dict[str, object]:
|
||||
manifest = load_manifest(pathlib.Path(args.source_manifest))
|
||||
binary = pathlib.Path(args.binary)
|
||||
repository_root = pathlib.Path(args.repository_root)
|
||||
database = pathlib.Path(args.database)
|
||||
for path in (binary, repository_root, database):
|
||||
if path.is_symlink():
|
||||
raise ValueError("shadow inputs cannot be symlinks")
|
||||
expected_heads = {
|
||||
str(item["path"]): str(item["head"]) for item in manifest["repositories"]
|
||||
}
|
||||
integrity, repository_count, user_count = database_readback(database)
|
||||
checks = {
|
||||
"binary_sha256": sha256(binary) == manifest.get("binary_sha256"),
|
||||
"repository_heads": repository_heads(repository_root) == expected_heads,
|
||||
"database_integrity": integrity == "ok",
|
||||
"database_repository_count": repository_count
|
||||
== manifest.get("database_repository_count"),
|
||||
"database_user_count": user_count == manifest.get("database_user_count"),
|
||||
"http_version": http_version(args.version_url) == manifest.get("forgejo_version"),
|
||||
}
|
||||
if not all(checks.values()):
|
||||
failed = ",".join(key for key, passed in checks.items() if not passed)
|
||||
raise ValueError(f"independent Forgejo shadow mismatch: {failed}")
|
||||
return {
|
||||
"schema": "guanghu.independent-forgejo-shadow-verification/v1",
|
||||
"result": "PASS_100",
|
||||
"source_node": manifest.get("source_node"),
|
||||
"shadow_node": args.shadow_node,
|
||||
"repository_count": len(expected_heads),
|
||||
"database_repository_count": repository_count,
|
||||
"database_user_count": user_count,
|
||||
"checks": checks,
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--source-manifest", required=True)
|
||||
parser.add_argument("--binary", required=True)
|
||||
parser.add_argument("--repository-root", required=True)
|
||||
parser.add_argument("--database", required=True)
|
||||
parser.add_argument("--version-url", required=True)
|
||||
parser.add_argument("--shadow-node", required=True)
|
||||
args = parser.parse_args()
|
||||
try:
|
||||
result = verify(args)
|
||||
except (OSError, ValueError, sqlite3.Error, subprocess.SubprocessError) as error:
|
||||
raise SystemExit(str(error)) from error
|
||||
print(json.dumps(result, sort_keys=True, separators=(",", ":")))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Reference in a new issue