ai-agent/backend/envfile.py

"""Order- and comment-preserving `.env` parsing/editing for project env files.

Vendored subset of ``services/account-manager/cli/envfile.py`` (the canonical
homelab env editor), extended with the **secrets section** convention the
project env editor exposes:

- A comment line matching ``# --- secrets ---`` (any of ``# secrets``,
  ``## Secrets``, ``# --- secret ---`` … — see ``SECRETS_MARKER_RE``) splits the
  file. Every assignment **below** the first marker is a *secret* var; everything
  above is a regular var.
- Secrets are never returned to the UI — only their keys and whether a value is
  set. They can be overwritten or removed, not read.

Unlike the account-manager copy, values here are stored **verbatim** (no
Compose ``$``→``$$`` doubling): ``~/projects/*`` env files are read by
Vite/dotenv/shells, not by Docker Compose ``env_file``.
"""
from __future__ import annotations

import os
import re
import tempfile
from dataclasses import dataclass, field

# A `.env` assignment line: optional leading `export`, KEY, `=`, then the rest.
_ASSIGN = re.compile(r"^\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=(.*)$")

# The comment line that opens the secrets section.
SECRETS_MARKER_RE = re.compile(r"^\s*#+\s*(?:-+\s*)?secrets?\b", re.IGNORECASE)
SECRETS_MARKER = "# --- secrets ---"

KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


def _needs_quoting(value: str) -> bool:
    if value == "":
        return False  # write `KEY=` bare
    if value != value.strip():
        return True  # leading/trailing whitespace
    return any(c in value for c in (" ", "\t", "#", '"', "'", "\n"))


def _quote(value: str) -> str:
    if not _needs_quoting(value):
        return value
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def _unquote(raw: str) -> str:
    """Turn the RHS of a `.env` line into its stored value."""
    raw = raw.strip()
    # Strip a trailing inline comment only when the value is unquoted. A bare
    # ` # ...` after an unquoted value is a comment; inside quotes it is data.
    if raw and raw[0] in ('"', "'"):
        quote = raw[0]
        i = 1
        out: list[str] = []
        while i < len(raw):
            c = raw[i]
            if c == "\\" and quote == '"' and i + 1 < len(raw):
                nxt = raw[i + 1]
                out.append({"n": "\n", "t": "\t", "r": "\r"}.get(nxt, nxt))
                i += 2
                continue
            if c == quote:
                break
            out.append(c)
            i += 1
        return "".join(out)
    if raw.startswith("#"):
        return ""
    hash_idx = raw.find(" #")
    if hash_idx != -1:
        raw = raw[:hash_idx]
    return raw.strip()


@dataclass
class ProjectEnvFile:
    path: str
    _lines: list[str] = field(default_factory=list)

    @classmethod
    def load(cls, path: str) -> "ProjectEnvFile":
        lines: list[str] = []
        if os.path.exists(path):
            with open(path, "r", encoding="utf-8") as fh:
                lines = fh.read().splitlines()
        return cls(path=path, _lines=lines)

    # --- sections ---------------------------------------------------------
    def _marker_index(self) -> int:
        """Line index of the first secrets marker, or -1."""
        for i, line in enumerate(self._lines):
            if SECRETS_MARKER_RE.match(line):
                return i
        return -1

    def _find(self, key: str) -> int:
        """Index of the last assignment line for `key`, or -1."""
        found = -1
        for i, line in enumerate(self._lines):
            m = _ASSIGN.match(line)
            if m and m.group(1) == key:
                found = i
        return found

    def entries(self) -> list[dict]:
        """Ordered vars: ``{key, value, secret}`` (last assignment wins)."""
        marker = self._marker_index()
        by_key: dict[str, dict] = {}
        for i, line in enumerate(self._lines):
            m = _ASSIGN.match(line)
            if not m:
                continue
            by_key[m.group(1)] = {
                "key": m.group(1),
                "value": _unquote(m.group(2)),
                "secret": marker != -1 and i > marker,
            }
        return list(by_key.values())

    # --- writing ----------------------------------------------------------
    def set_var(self, key: str, value: str, *, secret: bool) -> None:
        """Set `key`, placing a new key in the right section.

        An existing key in the right section is rewritten in place; one in the
        wrong section is removed and re-inserted where it belongs.
        """
        idx = self._find(key)
        if idx != -1:
            marker = self._marker_index()
            currently_secret = marker != -1 and idx > marker
            if currently_secret == secret:
                self._lines[idx] = f"{key}={_quote(value)}"
                return
            self.unset(key)

        new_line = f"{key}={_quote(value)}"
        marker = self._marker_index()
        if secret:
            if marker == -1:
                if self._lines and self._lines[-1].strip() != "":
                    self._lines.append("")
                self._lines.append(SECRETS_MARKER)
            self._lines.append(new_line)
        elif marker == -1:
            self._lines.append(new_line)
        else:
            # Insert above the marker, before its preceding blank separator.
            at = marker
            while at > 0 and self._lines[at - 1].strip() == "":
                at -= 1
            self._lines.insert(at, new_line)

    def unset(self, key: str) -> bool:
        removed = False
        kept: list[str] = []
        for line in self._lines:
            m = _ASSIGN.match(line)
            if m and m.group(1) == key:
                removed = True
                continue
            kept.append(line)
        self._lines = kept
        return removed

    def save(self) -> None:
        body = "\n".join(self._lines)
        if body and not body.endswith("\n"):
            body += "\n"
        d = os.path.dirname(os.path.abspath(self.path)) or "."
        fd, tmp = tempfile.mkstemp(dir=d, prefix=".env.", suffix=".tmp")
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as fh:
                fh.write(body)
            os.replace(tmp, self.path)
            try:
                os.chmod(self.path, 0o600)
            except OSError:
                pass
        finally:
            if os.path.exists(tmp):
                os.unlink(tmp)