monky-deployd/monky_deployd/state.py

"""On-disk state under /var/lib/monky-deployd (0700): state.json, bao.token, the flock.

Everything is written atomically (tmp + fsync + rename) with mode 0600; the token file is the
ONLY credential the agent keeps, and it is the agent's bearer to tenancy (ADR-0028 §C.C)."""

from __future__ import annotations

import fcntl
import json
import os
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path

HISTORY_MAX = 20


@dataclass
class TokenMeta:
    accessor: str | None = None
    issued_at: float | None = None
    ttl_s: int | None = None
    renewable: bool | None = None
    grant_jti: str | None = None
    source: str | None = None  # bootstrap | lease


@dataclass
class State:
    env_id: str
    applied_sha: str | None = None
    applied_at: float | None = None
    last_checkin_at: float | None = None
    last_report_at: float | None = None
    last_result: str | None = None
    last_error: str | None = None
    last_action: str | None = None
    desired_sha: str | None = None
    history: list[str] = field(default_factory=list)
    token: TokenMeta = field(default_factory=TokenMeta)
    agent_version: str | None = None
    updated_at: float | None = None
    consecutive_failures: int = 0

    def remember_applied(self, sha: str) -> None:
        self.applied_sha = sha
        self.applied_at = time.time()
        if sha in self.history:
            self.history.remove(sha)
        self.history.append(sha)
        del self.history[:-HISTORY_MAX]

    def is_rollback(self, sha: str) -> bool:
        """`sha` was applied before AND something newer has been applied since."""
        if sha not in self.history:
            return False
        return self.history.index(sha) < len(self.history) - 1


def write_private(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        with os.fdopen(fd, "wb") as fh:
            fh.write(data)
            fh.flush()
            os.fsync(fh.fileno())
        os.chmod(tmp, 0o600)
        os.replace(tmp, path)
    except BaseException:
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass
        raise


def load(path: Path, env_id: str) -> State:
    try:
        raw = json.loads(path.read_text())
    except FileNotFoundError:
        return State(env_id=env_id)
    except (OSError, json.JSONDecodeError):
        return State(env_id=env_id, last_error="state.json unreadable; reset")
    tok = raw.pop("token", None) or {}
    known = {k for k in State.__dataclass_fields__}
    st = State(**{k: v for k, v in raw.items() if k in known and k != "token"})
    st.token = TokenMeta(**{k: v for k, v in tok.items() if k in TokenMeta.__dataclass_fields__})
    if st.env_id != env_id:
        # state belongs to another env: never mix histories
        return State(env_id=env_id, last_error=f"state.json was for {st.env_id}; reset")
    return st


def save(path: Path, st: State) -> None:
    st.updated_at = time.time()
    write_private(path, (json.dumps(asdict(st), indent=2, sort_keys=True) + "\n").encode())


def read_token(path: Path) -> str | None:
    try:
        tok = path.read_text().strip()
    except FileNotFoundError:
        return None
    return tok or None


def write_token(path: Path, token: str) -> None:
    write_private(path, token.encode())


def delete(path: Path) -> bool:
    """Remove a credential file. If the directory is not ours to write (/etc/monky-deployd is
    root-owned), truncating the file to zero bytes is just as final: an empty grant is no grant."""
    try:
        path.unlink()
        return True
    except FileNotFoundError:
        return False
    except PermissionError:
        try:
            with open(path, "wb"):
                pass
            return True
        except OSError:
            return False


class Lock:
    """Non-blocking flock; a second concurrent tick exits quietly (the timer will fire again)."""

    def __init__(self, path: Path):
        self.path = path
        self._fd: int | None = None

    def acquire(self) -> bool:
        self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        self._fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o600)
        try:
            fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            os.close(self._fd)
            self._fd = None
            return False
        os.write(self._fd, str(os.getpid()).encode())
        return True

    def release(self) -> None:
        if self._fd is not None:
            try:
                fcntl.flock(self._fd, fcntl.LOCK_UN)
            finally:
                os.close(self._fd)
                self._fd = None