credentials.py

Go to the documentation of this file.

"""Sidecar-backed credential broker with a runtime gitignore canary.
 
The implementation follows the plan-216 provider sidecar pattern. Raw values are
accepted by the backend and returned only to the caller that asks for a named
credential. Diagnostics, printed output, logging, and exception messages carry
names and paths only.
 
Dependencies
------------
- oc_connect.workspace (credential_path - the default workspace-anchored sidecar path)
- oc_diagnostics (_dbg structured logger)
- git (not imported; run as git check-ignore, the canary that proves the sidecar is gitignored)
"""
from __future__ import annotations
 
from contextlib import contextmanager
import json
import os
import re
import secrets as _secrets
import stat
import subprocess
from collections.abc import Callable, Iterator, Mapping
from pathlib import Path
from typing import Protocol
 
from oc_diagnostics import _dbg
 
from .workspace import credential_path
 
SECRET_FILE_MODE = 0o600
_NAME_RE = re.compile(r"^[a-z][a-z0-9_.-]*$")
_GIT_TIMEOUT_SECONDS = 5.0
 
 
class CredentialPathError(ValueError):
    """The selected credential file cannot be proven gitignored."""
 
 
class CredentialFileError(ValueError):
    """The credential file is unreadable or malformed."""
 
 
class CredentialNotFoundError(LookupError):
    """A requested credential name has no stored value."""
 
 
class CredentialBackend(Protocol):
    """Storage boundary hidden behind :class:`CredentialBroker`."""
 
    def get(self, name: str) -> str | None: ...
 
    def set(self, name: str, value: str) -> None: ...
 
    def delete(self, name: str) -> None: ...
 
    def snapshot(self) -> bytes | None: ...
 
    def restore(self, snapshot: bytes | None) -> None: ...
 
 
def _existing_ancestor(path: Path) -> Path:
    candidate = path if path.is_dir() else path.parent
    while not candidate.is_dir() and candidate != candidate.parent:
        candidate = candidate.parent
    return candidate
 
 
def is_gitignored(path: str | Path) -> bool:
    """Ask git whether *path* is ignored; every probe error fails closed."""
    target = Path(path).expanduser()
    try:
        target = target.resolve()
    except OSError:
        target = target.absolute()
    try:
        result = subprocess.run(
            ["git", "check-ignore", "--quiet", "--", str(target)],
            cwd=str(_existing_ancestor(target)),
            capture_output=True,
            text=True,
            timeout=_GIT_TIMEOUT_SECONDS,
            check=False,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        _dbg(
            "OC_CONNECT.CREDENTIALS",
            "is_gitignored",
            f"SECURITY_ERROR: gitignore probe failed for {target}: {type(exc).__name__}",
            1,
            override=True,
        )
        return False
    covered = result.returncode == 0
    _dbg(
        "OC_CONNECT.CREDENTIALS",
        "is_gitignored",
        f"gitignore canary path={target} covered={covered} rc={result.returncode}",
        3,
    )
    return covered
 
 
def _chmod_private(path: Path) -> None:
    try:
        path.chmod(stat.S_IRUSR | stat.S_IWUSR)
    except OSError:
        pass
 
 
class SidecarCredentialBackend:
    """Atomic JSON sidecar backend guarded by a live git check-ignore probe."""
 
    def __init__(
        self,
        path: str | Path | None = None,
        *,
        gitignore_check: Callable[[str | Path], bool] = is_gitignored,
        root: str | Path | None = None,
    ) -> None:
        self.path = Path(path) if path is not None else credential_path(root)
        self._gitignore_check = gitignore_check
 
    def _assert_safe_path(self) -> None:
        if not self._gitignore_check(self.path):
            raise CredentialPathError(
                f"refusing to use connector credentials at {self.path}: the path is not "
                "covered by gitignore (checked with 'git check-ignore')"
            )
 
    def _load(self) -> dict[str, str]:
        if not self.path.is_file():
            return {}
        self._assert_safe_path()
        try:
            value = json.loads(self.path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise CredentialFileError(
                f"could not parse connector credential sidecar at {self.path} "
                f"({type(exc).__name__})"
            ) from None
        if not isinstance(value, Mapping) or not all(
            isinstance(name, str) and isinstance(item, str)
            for name, item in value.items()
        ):
            raise CredentialFileError(
                f"connector credential sidecar at {self.path} must be a string map"
            )
        return dict(value)
 
    def _write(self, values: Mapping[str, str]) -> None:
        self._assert_safe_path()
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary = self.path.with_name(
            f".tmp-{self.path.name}-{_secrets.token_hex(8)}"
        )
        try:
            with temporary.open("w", encoding="utf-8", newline="\n") as handle:
                json.dump(dict(sorted(values.items())), handle, indent=2, sort_keys=True)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
            _chmod_private(temporary)
            os.replace(temporary, self.path)
            _chmod_private(self.path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
    def get(self, name: str) -> str | None:
        """Return the raw value for *name*, or ``None`` when absent."""
        return self._load().get(name)
 
    def set(self, name: str, value: str) -> None:
        """Store one value without exposing it to diagnostics."""
        values = self._load()
        values[name] = value
        self._write(values)
 
    def delete(self, name: str) -> None:
        """Delete one value; absent names are idempotent."""
        values = self._load()
        if name not in values:
            return
        del values[name]
        self._write(values)
 
    def snapshot(self) -> bytes | None:
        """Return the exact sidecar bytes for an internal transaction rollback."""
        if not self.path.is_file():
            return None
        self._assert_safe_path()
        try:
            return self.path.read_bytes()
        except OSError as exc:
            raise CredentialFileError(
                f"could not snapshot connector credential sidecar at {self.path} "
                f"({type(exc).__name__})"
            ) from None
 
    def restore(self, snapshot: bytes | None) -> None:
        """Restore an internal snapshot without exposing its bytes."""
        self._assert_safe_path()
        if snapshot is None:
            try:
                self.path.unlink(missing_ok=True)
            except OSError as exc:
                raise CredentialFileError(
                    f"could not restore connector credential sidecar at {self.path} "
                    f"({type(exc).__name__})"
                ) from None
            return
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary = self.path.with_name(
            f".tmp-{self.path.name}-{_secrets.token_hex(8)}"
        )
        try:
            with temporary.open("wb") as handle:
                handle.write(snapshot)
                handle.flush()
                os.fsync(handle.fileno())
            _chmod_private(temporary)
            os.replace(temporary, self.path)
            _chmod_private(self.path)
        finally:
            if temporary.exists():
                try:
                    temporary.unlink()
                except OSError:
                    pass
 
 
class CredentialBroker:
    """Name-only application boundary around a credential backend."""
 
    def __init__(
        self,
        backend: CredentialBackend | None = None,
        *,
        root: str | Path | None = None,
    ) -> None:
        self.backend = backend or SidecarCredentialBackend(root=root)
 
    @contextmanager
    def transaction(self) -> Iterator[None]:
        """Roll the backend back byte-for-byte if the guarded flow fails."""
        snapshot = self.backend.snapshot()
        try:
            yield
        except BaseException:
            self.backend.restore(snapshot)
            _dbg(
                "OC_CONNECT.CREDENTIALS",
                "transaction",
                "credential transaction rolled back",
                2,
            )
            raise
 
    @staticmethod
    def _validate_name(name: str) -> str:
        if not isinstance(name, str) or _NAME_RE.fullmatch(name) is None:
            raise ValueError(
                "credential name must start with a lowercase letter and contain only "
                "lowercase letters, digits, dots, dashes, or underscores"
            )
        return name
 
    def store(self, name: str, value: str) -> None:
        """Store one named value; diagnostics record only the name."""
        checked_name = self._validate_name(name)
        if not isinstance(value, str) or not value.strip():
            raise ValueError(f"refusing to store an empty value for {checked_name}")
        self.backend.set(checked_name, value.strip())
        _dbg(
            "OC_CONNECT.CREDENTIALS",
            "store",
            f"stored connector credential name={checked_name}",
            2,
        )
 
    def retrieve(self, name: str) -> str:
        """Return one raw value to its transport caller and nowhere else."""
        checked_name = self._validate_name(name)
        value = self.backend.get(checked_name)
        if value is None:
            raise CredentialNotFoundError(
                f"connector credential {checked_name!r} is not configured"
            )
        _dbg(
            "OC_CONNECT.CREDENTIALS",
            "retrieve",
            f"retrieved connector credential name={checked_name}",
            3,
        )
        return value
 
    def delete(self, name: str) -> None:
        """Delete one named value without exposing its former content."""
        checked_name = self._validate_name(name)
        self.backend.delete(checked_name)
        _dbg(
            "OC_CONNECT.CREDENTIALS",
            "delete",
            f"deleted connector credential name={checked_name}",
            2,
        )
 
 
__all__ = [
    "CredentialBackend",
    "CredentialBroker",
    "CredentialFileError",
    "CredentialNotFoundError",
    "CredentialPathError",
    "SidecarCredentialBackend",
    "is_gitignored",
]