guanghu-ice-heart/server-tools/enterprise-lighthouse/domain_manifest_signing.py

131 lines
4.5 KiB
Python

#!/usr/bin/env python3
"""Disabled-by-default Ed25519 envelope signer for HoloLake domain manifests.
This module defines source bytes shared with the HoloLake client. It does not
register a production signer, generate a key, expose an HTTP route, or claim
that a domain runtime is online.
"""
import base64
import hashlib
import json
import os
import re
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping, Optional
MANIFEST_SCHEMA = "gh-aios.domain-manifest/v1"
IDENTIFIER = re.compile(r"^[A-Z0-9][A-Z0-9._:-]{0,159}$")
SOURCE_COMMIT = re.compile(r"^[a-f0-9]{40}(?:[a-f0-9]{24})?$")
class SignerConfigurationError(ValueError):
pass
@dataclass(frozen=True)
class SignerConfig:
private_key_path: Path
repository_id: str
signer_id: str
def _identifier(value: str, field: str) -> str:
if not isinstance(value, str) or not IDENTIFIER.fullmatch(value):
raise SignerConfigurationError(f"{field} is not a bounded identifier")
return value
def canonical_manifest_payload(*, domain_id: str, repository_id: str, signer_id: str, source_commit: str) -> bytes:
_identifier(domain_id, "domain_id")
_identifier(repository_id, "repository_id")
_identifier(signer_id, "signer_id")
if not isinstance(source_commit, str) or not SOURCE_COMMIT.fullmatch(source_commit):
raise SignerConfigurationError("source_commit must be a full Git object id")
payload = {
"domainId": domain_id,
"repositoryId": repository_id,
"schema": MANIFEST_SCHEMA,
"signerId": signer_id,
"sourceCommit": source_commit,
}
return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def _private_key(path: Path) -> Path:
if not isinstance(path, Path) or not path.is_absolute():
raise SignerConfigurationError("private key path must be absolute")
try:
status = path.stat()
except OSError as error:
raise SignerConfigurationError("private key file is unavailable") from error
if not path.is_file() or status.st_mode & 0o077:
raise SignerConfigurationError("private key file must be regular and owner-only")
return path
def load_signer_config_from_environment(environment: Mapping[str, str] = os.environ) -> Optional[SignerConfig]:
key_path = environment.get("LIGHTHOUSE_DOMAIN_SIGNING_KEY_PATH", "").strip()
signer_id = environment.get("LIGHTHOUSE_DOMAIN_SIGNER_ID", "").strip()
repository_id = environment.get("LIGHTHOUSE_DOMAIN_SIGNING_REPOSITORY_ID", "").strip()
if not key_path and not signer_id and not repository_id:
return None
if not key_path or not signer_id or not repository_id:
raise SignerConfigurationError("domain signing configuration is incomplete")
return SignerConfig(
private_key_path=_private_key(Path(key_path)),
repository_id=_identifier(repository_id, "repository_id"),
signer_id=_identifier(signer_id, "signer_id"),
)
def sign_domain_manifest(
*,
private_key_path: Path,
domain_id: str,
repository_id: str,
signer_id: str,
source_commit: str,
) -> dict:
key_path = _private_key(private_key_path)
payload = canonical_manifest_payload(
domain_id=domain_id,
repository_id=repository_id,
signer_id=signer_id,
source_commit=source_commit,
)
payload_path = None
try:
with tempfile.NamedTemporaryFile(prefix="guanghu-domain-manifest-", delete=False) as handle:
handle.write(payload)
payload_path = Path(handle.name)
payload_path.chmod(0o600)
completed = subprocess.run(
[
"openssl", "pkeyutl", "-sign", "-rawin", "-inkey", str(key_path),
"-in", str(payload_path),
],
capture_output=True,
check=True,
timeout=5,
)
except (OSError, subprocess.SubprocessError) as error:
raise SignerConfigurationError("Ed25519 manifest signing failed") from error
finally:
if payload_path is not None:
payload_path.unlink(missing_ok=True)
if len(completed.stdout) != 64:
raise SignerConfigurationError("signer returned an invalid Ed25519 signature")
return {
"digest": hashlib.sha256(payload).hexdigest(),
"domainId": domain_id,
"repositoryId": repository_id,
"schema": MANIFEST_SCHEMA,
"signature": base64.b64encode(completed.stdout).decode("ascii"),
"signerId": signer_id,
"sourceCommit": source_commit,
}