99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import json
|
|
from typing import Protocol
|
|
|
|
from .models import ProtectionSnapshot
|
|
|
|
|
|
class CASConflict(RuntimeError):
|
|
pass
|
|
|
|
|
|
class DormantViolation(RuntimeError):
|
|
pass
|
|
|
|
|
|
def snapshot_sha256(snapshot: ProtectionSnapshot) -> str:
|
|
raw = json.dumps(snapshot.as_dict(), sort_keys=True, separators=(",", ":")).encode()
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
class ProtectionStore(Protocol):
|
|
fixture: bool
|
|
|
|
def read(self, branch: str) -> ProtectionSnapshot: ...
|
|
def compare_and_swap(
|
|
self, branch: str, expected_sha256: str, target: ProtectionSnapshot
|
|
) -> bool: ...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProtectionPlan:
|
|
branch: str
|
|
expected_previous_sha256: str
|
|
previous: ProtectionSnapshot
|
|
target: ProtectionSnapshot
|
|
target_sha256: str
|
|
|
|
|
|
class BranchProtectionCASMigrationAdapter:
|
|
"""Candidate adapter. Mutations are restricted to disposable fixtures."""
|
|
|
|
def plan(
|
|
self,
|
|
current: ProtectionSnapshot,
|
|
target: ProtectionSnapshot,
|
|
expected_previous_sha256: str,
|
|
) -> ProtectionPlan:
|
|
actual = snapshot_sha256(current)
|
|
if actual != expected_previous_sha256:
|
|
raise CASConflict("EXPECTED_PREVIOUS_SHA256_MISMATCH")
|
|
if current.branch != target.branch:
|
|
raise CASConflict("BRANCH_BINDING_MISMATCH")
|
|
if target.direct_push or target.force_push or target.branch_deletion:
|
|
raise DormantViolation("UNSAFE_TARGET_POLICY")
|
|
return ProtectionPlan(
|
|
current.branch, actual, current, target, snapshot_sha256(target)
|
|
)
|
|
|
|
def apply_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
|
|
if not store.fixture:
|
|
raise DormantViolation("LIVE_PROTECTION_MUTATION_DENIED")
|
|
if not store.compare_and_swap(
|
|
plan.branch, plan.expected_previous_sha256, plan.target
|
|
):
|
|
raise CASConflict("CONCURRENT_PROTECTION_CHANGE")
|
|
return store.read(plan.branch)
|
|
|
|
def rollback_fixture(self, store: ProtectionStore, plan: ProtectionPlan) -> ProtectionSnapshot:
|
|
if not store.fixture:
|
|
raise DormantViolation("LIVE_PROTECTION_ROLLBACK_DENIED")
|
|
if not store.compare_and_swap(plan.branch, plan.target_sha256, plan.previous):
|
|
raise CASConflict("ROLLBACK_CONFLICT")
|
|
return store.read(plan.branch)
|
|
|
|
|
|
class MemoryProtectionStore:
|
|
fixture = True
|
|
|
|
def __init__(self, snapshot: ProtectionSnapshot) -> None:
|
|
self._snapshot = snapshot
|
|
|
|
def read(self, branch: str) -> ProtectionSnapshot:
|
|
if self._snapshot.branch != branch:
|
|
raise KeyError(branch)
|
|
return self._snapshot
|
|
|
|
def compare_and_swap(
|
|
self, branch: str, expected_sha256: str, target: ProtectionSnapshot
|
|
) -> bool:
|
|
if self._snapshot.branch != branch:
|
|
return False
|
|
if snapshot_sha256(self._snapshot) != expected_sha256:
|
|
return False
|
|
self._snapshot = target
|
|
return True
|