=========== __init__
=========== feature
from controllers.base import SAVE_REWRITES
from features.base import Feature
from features.journal import Journal
from features.secrets.controller import Secrets
from features.secrets.details import SecretsDetails
from features.secrets.guard import RefuseSecretsFile
from features.secrets.handlers import PurgeDeletedSecrets
from features.secrets.leaks import LeakAlarm

__all__ = ["Secrets"]


class SecretsFeature(Feature):
    details = SecretsDetails

    def register(self, journal: Journal) -> None:
        journal.events.handler(PurgeDeletedSecrets())
        journal.agent.interceptor(RefuseSecretsFile())
        SAVE_REWRITES.add(self, LeakAlarm())
=========== resource
from dataclasses import asdict, dataclass
from enum import StrEnum
from typing import ClassVar

from engine.wording import slugged
from resources.base import PROJECT, UNLISTED, Refused, Resource, ResourceDetails
from resources.shapes import Field, Shape


@dataclass(frozen=True)
class SecretField:
    name: str
    hidden: bool
    variable: str

    @classmethod
    def from_json(cls, raw: dict) -> "SecretField":
        return cls(raw["name"], bool(raw.get("hidden", True)), raw["variable"])


class Kind(StrEnum):
    API_KEY = "api key"
    LOGIN = "login"
    CUSTOM = "custom"

    @classmethod
    def named(cls, given: str) -> "Kind":
        try:
            return cls(given.strip().lower())
        except ValueError as missing:
            raise Refused(f"a secret is one of {', '.join(kind.value for kind in cls)}, not {given!r}") from missing

    def fields(self, title: str) -> list[dict]:
        stem = slugged(title, "_").upper()
        names = {Kind.API_KEY: (("key", True),), Kind.LOGIN: (("username", False), ("password", True))}.get(self, ())
        return [asdict(SecretField(name, hidden, f"{stem}_{name.upper()}")) for name, hidden in names]


class Secret(Shape, Resource):
    details: ClassVar[ResourceDetails] = ResourceDetails(
        title="Secret",
        abstract="A key or login the agent may use without ever seeing it",
        help="Its values live in a file in your home folder, never in the journal. You fill them in under Settings, Secrets.",
    )
    data_fields: ClassVar[list[Field]] = [
        Field(default=Kind.CUSTOM.value, name="kind"),
        Field(default=list, name="secret_fields"),
        Field(default=list, name="programs"),
        Field(default=False, name="helpers"),
        Field(default=dict, name="filled", journal_only=True),
        Field(default=0.0, name="used", journal_only=True),
        Field(default="", name="asked"),
        Field(default=0.0, name="session", journal_only=True),
        Field(default=0.0, name="session_expires", journal_only=True),
    ]
    type = "secret"
    icon = "key"
    scope = PROJECT
    listed_under = UNLISTED
    in_sidebar = False
    subagent_writable = False
    takes_comments = False

    def field(self, name: str) -> SecretField:
        found = next((SecretField.from_json(raw) for raw in self.secret_fields if raw["name"] == name), None)
        if found is None:
            raise Refused(f"secret {self.n} has no field {name!r}; its fields are {', '.join(raw['name'] for raw in self.secret_fields) or 'none yet'}")
        return found

    def is_waiting(self) -> bool:
        return any(raw["name"] not in self.filled for raw in self.secret_fields)
=========== values
import json
import os
import uuid
from functools import cache
from pathlib import Path

PROJECT_ID = "project-id"
FOLDER_VARIABLE = "AGENT_JOURNAL_SECRETS"
HEADER = """# The values of this project's secrets, written by agent-journal.
# One line per field: NAME="value". The journal never copies this file anywhere.
"""


def secrets_folder() -> Path:
    if FOLDER_VARIABLE in os.environ:
        return Path(os.environ[FOLDER_VARIABLE])
    if os.name == "nt":
        return Path(os.environ["APPDATA"]) / "agent-journal" / "secrets"
    return Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "agent-journal" / "secrets"


@cache
def project_id(root: Path) -> str:
    path = root / PROJECT_ID
    if not path.is_file():
        path.write_text(uuid.uuid4().hex)
    return path.read_text().strip()


def parsed(line: str) -> tuple[str, str] | None:
    name, equals, rest = line.strip().partition("=")
    if not equals or name.startswith("#"):
        return None
    return name.strip(), json.loads(rest) if rest.startswith('"') else rest


class ValuesFile:
    def __init__(self, root: Path):
        self.path = secrets_folder() / f"{project_id(root)}.env"

    def values(self) -> dict[str, str]:
        if not self.path.is_file():
            return {}
        return dict(found for found in map(parsed, self.path.read_text().splitlines()) if found)

    def put(self, variable: str, value: str) -> None:
        self.write({**self.values(), variable: value})

    def drop(self, variables: list[str]) -> None:
        self.write({name: value for name, value in self.values().items() if name not in variables})

    def write(self, values: dict[str, str]) -> None:
        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        self.path.parent.chmod(0o700)
        written = self.path.with_suffix(".writing")
        handle = os.open(written, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(handle, "w") as out:
            out.write(HEADER + "".join(f"{name}={json.dumps(value)}\n" for name, value in values.items()))
        os.replace(written, self.path)
=========== running
import base64
import contextlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import urllib.parse
from pathlib import Path

from resources.base import Refused

CHUNK = 4096
NEVER_GIVEN = re.compile(r"^(?:a?sh|bash|zsh|fish|dash|ksh|t?csh|env|xargs|sudo|su|doas|nohup|nice|time|timeout|watch|script|eval|exec|"
                         r"python[\d.]*|pypy[\d.]*|node|deno|bun|ruby|perl[\d.]*|php[\d.]*|lua[\d.]*|osascript|"
                         r"make|npm|npx|yarn|pnpm|pip[\d.]*|uv|uvx|cargo|go|java|dotnet|gradle|mvn|composer)$")
THROWAWAY = ("HOME", "XDG_CONFIG_HOME", "XDG_CACHE_HOME", "XDG_DATA_HOME", "XDG_STATE_HOME", "GH_CONFIG_DIR", "DOCKER_CONFIG")


def forms(value: str) -> list[bytes]:
    raw = value.encode()
    found = {raw, base64.b64encode(raw), base64.urlsafe_b64encode(raw), urllib.parse.quote(value, safe="").encode(),
             json.dumps(value)[1:-1].encode(), raw.hex().encode()}
    return sorted((form.rstrip(b"=") for form in found if form), key=len, reverse=True)


class Masker:
    def __init__(self, masks: dict[str, str]):
        self.masks = [(form, f"[secret {name}]".encode()) for value, name in masks.items() for form in forms(value)]
        self.carried = max((len(form) for form, _ in self.masks), default=1) - 1
        self.held = b""

    def feed(self, chunk: bytes) -> bytes:
        self.held = self.masked(self.held + chunk)
        cut = max(len(self.held) - self.carried, 0)
        ready, self.held = self.held[:cut], self.held[cut:]
        return ready

    def flush(self) -> bytes:
        ready, self.held = self.masked(self.held), b""
        return ready

    def masked(self, text: bytes) -> bytes:
        for form, mask in self.masks:
            text = text.replace(form, mask)
        return text


def checked_program(command: tuple[str, ...], allowed: list[str]) -> str:
    if not command:
        raise Refused("name the command after --, such as journal secret run github -- gh api user")
    program = Path(command[0]).name
    if NEVER_GIVEN.match(program):
        raise Refused(f"a secret is never given to {program}: name the program that uses it directly, such as curl or gh")
    if allowed and program not in allowed:
        raise Refused(f"this secret may be given only to {', '.join(allowed)}, not {program}")
    if shutil.which(command[0]) is None:
        raise Refused(f"no program {command[0]} here")
    return program


def run_masked(command: tuple[str, ...], values: dict[str, str], masks: dict[str, str], given: bytes | None) -> int:
    home = tempfile.mkdtemp(prefix="journal-secret-")
    environment = {**os.environ, **dict.fromkeys(THROWAWAY, home), **values}
    masker = Masker(masks)
    try:
        child = subprocess.Popen(command, stdin=subprocess.PIPE if given is not None else subprocess.DEVNULL,
                                 stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=environment)
        if given is not None:
            with contextlib.suppress(BrokenPipeError), child.stdin as stdin:  # a program that never reads its input may exit first
                stdin.write(given)
        out = sys.stdout.buffer
        for chunk in iter(lambda: child.stdout.read1(CHUNK), b""):
            out.write(masker.feed(chunk))
            out.flush()
        out.write(masker.flush())
        out.flush()
        return child.wait()
    finally:
        shutil.rmtree(home, ignore_errors=True)
=========== guard
from pathlib import Path

from engine.reach import Reach
from features.parts import AgentContext, ToolInterceptor
from features.secrets.values import secrets_folder

NAMED = ("agent-journal/secrets", "agent-journal\\secrets")


class RefuseSecretsFile(ToolInterceptor):
    reach = Reach.BOTH

    def intercept(self, context: AgentContext, call) -> str:
        folder = secrets_folder().resolve()
        here = Path(context.hook.cwd) if context.hook.cwd else Path.cwd()
        read = any((here / path).resolve().is_relative_to(folder) for path in call.paths)
        named = any(word in command for command in call.commands for word in (*NAMED, str(folder)))
        if not read and not named:
            return ""
        return "The secrets file is never read by an agent: use journal secret run <name> -- <command>, and ask the user for what is missing with journal secret request."
=========== leaks
import time
from pathlib import Path
from typing import Callable

from controllers.notifications import Notifications
from features.secrets.resource import SecretField
from features.secrets.values import ValuesFile
from resources.base import SYSTEM

SHORTEST = 8
TOLD_EVERY = 600


class LeakAlarm:
    known: dict[str, tuple[float, dict[str, str]]] = {}
    told: dict[tuple[str, str], float] = {}

    def __call__(self, record) -> Callable[[str], str]:
        names = self.names(record)
        return lambda text: self.masked(record, names, text) if text and isinstance(text, str) else text

    def names(self, record) -> dict[str, str]:
        path = ValuesFile(record.root).path
        stamp = path.stat().st_mtime if path.is_file() else 0.0
        held = self.known.get(str(path))
        if held is None or held[0] != stamp:
            held = (stamp, self.loaded(record))
            self.known[str(path)] = held
        return held[1]

    def loaded(self, record) -> dict[str, str]:
        from features.secrets.controller import Secrets
        values = ValuesFile(record.root).values()
        fields = ((row.title, SecretField.from_json(raw)) for row in Secrets(record, actor=SYSTEM).rows.every() for raw in row.secret_fields)
        return {values[field.variable]: title for title, field in fields if field.hidden and len(values.get(field.variable, "")) >= SHORTEST}

    def masked(self, record, names: dict[str, str], text: str) -> str:
        for value, title in names.items():
            if value in text:
                text = text.replace(value, f"[secret {title}]")
                self.tell(record, title)
        return text

    def tell(self, record, title: str) -> None:
        key = (str(Path(record.root)), title)
        if time.time() - self.told.get(key, 0.0) < TOLD_EVERY:
            return
        self.told[key] = time.time()
        Notifications(record, actor=SYSTEM).create(f"The value of the secret {title} was written into the journal",
                                                   brief=f"It was replaced with [secret {title}] before it was saved, but it was seen: rotate it, then fill in the new value under Settings, Secrets.")
=========== handlers
import time

from engine.events.engine import ClockTicked
from features.parts import WHOLE_FEATURE, AgentContext, Handler
from features.secrets.controller import Secrets
from resources.base import SYSTEM

PURGE_EVERY = 3600


class PurgeDeletedSecrets(Handler):
    behaviour = WHOLE_FEATURE
    purged: dict[str, float] = {}

    def handle(self, context: AgentContext, event: ClockTicked) -> None:
        place = str(context.record.root)
        if time.time() - self.purged.get(place, 0.0) < PURGE_EVERY:
            return
        self.purged[place] = time.time()
        Secrets(context.record, actor=SYSTEM)._purge()
=========== sessions
import json
import os
import shutil
import subprocess
from pathlib import Path

from engine.wording import slugged
from features.secrets.values import project_id, secrets_folder
from resources.base import Refused

OPEN_BROWSER = ("npx", "-y", "playwright@latest", "open")


class BrowserLogins:
    def __init__(self, root: Path):
        self.folder = secrets_folder() / f"{project_id(root)}.sessions"
        self.merged = secrets_folder() / f"{project_id(root)}.browser.json"

    def saved_for(self, title: str) -> Path:
        return self.folder / f"{slugged(title)}.json"

    def record(self, title: str, url: str) -> Path:
        if shutil.which(OPEN_BROWSER[0]) is None:
            raise Refused("logging in needs npx (Node.js) on this machine")
        self.folder.mkdir(mode=0o700, parents=True, exist_ok=True)
        saved = self.saved_for(title)
        subprocess.run([*OPEN_BROWSER, f"--save-storage={saved}", url], check=False)
        if not saved.is_file():
            raise Refused("no login was saved: log in in the browser that opens, then close its window")
        saved.chmod(0o600)
        return saved

    def merge(self) -> Path:
        cookies, origins = {}, {}
        for saved in sorted(self.folder.glob("*.json")):
            state = json.loads(saved.read_text())
            cookies.update({(c["name"], c["domain"], c["path"]): c for c in state.get("cookies", [])})
            origins.update({o["origin"]: o for o in state.get("origins", [])})
        handle = os.open(self.merged, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(handle, "w") as out:
            json.dump({"cookies": list(cookies.values()), "origins": list(origins.values())}, out)
        return self.merged

    def expires(self, saved: Path) -> float:
        ends = [c["expires"] for c in json.loads(saved.read_text()).get("cookies", []) if c.get("expires", -1) > 0]
        return min(ends, default=0.0)
=========== details
from features.base import FeatureDetails
from features.groups import Group


class SecretsDetails(FeatureDetails):
    explains = "Keys and logins the agent may use without ever seeing them. You fill in their values under Settings, Secrets."
    name = "secrets"
    group = Group.PROJECT
    label = "Let the agent use keys and logins without seeing them"
    has_skill = True

    title = "Secrets"

    abstract = """
        A secret names a key or a login, says what it is for and how to use it, and lists the
        commands it may be given to. Its values live in a file in your home folder, outside the
        project, and never in the journal.
    """

    help = """
        When a task needs a key or a login you do not have, ask for it with journal secret request
        "<name>" "<why>" [--kind "api key"|login|custom]: it waits under Settings, Secrets until the
        user fills it in. Never ask the user to paste a value into the chat, and never read the
        file that holds the values.

        Use a secret with journal secret run <name> -- <command>: the command gets the hidden
        value on its standard input, a home folder of its own so it keeps no login behind, and
        its output comes back with every form of the value replaced by [secret <name>]. --stdin
        "<text>" sends that text instead, with {<field>} filled in, such as --stdin
        "Authorization: Bearer {key}" for curl -H @-; --env gives the fields as environment
        variables instead. A secret never goes to a shell, an interpreter or a build tool, only
        to the programs it lists, and to helpers and subagents only when the user shared it. journal secret all lists the secrets with their descriptions,
        instructions and fields; journal secret read <n> shows one. A value you made yourself,
        such as a generated password, is written to a file and moved into the secret with
        journal secret store <n> <field> <file>, which deletes the file.

        The values live in one file per project under your home folder, owner-only, never in git,
        the attic, a worktree or a backup of the record; journal secret where prints its path.
        A deleted secret keeps its values for 30 days, then they are removed from the file.

        For a site you use in a browser, never type a password into a page: run journal secret
        login <name> <url>, and the user logs in once in the browser that opens and closes it.
        The session is saved beside the values file, and the agent's own browser tool starts
        logged in from its next start.
    """
=========== controller
import re
import time
from pathlib import Path

import controllers.types as types_module
import resources.types as resources_module
from controllers.base import Controller
from controllers.marks import action
from controllers.types import Environments, Messages
from features.secrets.resource import Kind, Secret, SecretField
from features.secrets.running import checked_program, run_masked
from features.secrets.sessions import BrowserLogins
from features.secrets.values import ValuesFile
from providers import PROVIDERS
from resources.base import AGENT, SYSTEM, USER, Refused

KEPT_DAYS = 30
PLACEHOLDER = re.compile(r"\{(\w+)\}")


class Secrets(Controller):
    resource = Secret

    @action
    def create(self, title: str, abstract: str = "", brief: str = "", kind: str = Kind.CUSTOM.value, **data) -> Secret:
        chosen = Kind.named(kind)
        return super().create(title, abstract, brief, kind=chosen.value, **{"secret_fields": chosen.fields(title), **data})

    @action
    def request(self, title: str, why: str, kind: str = Kind.API_KEY.value) -> Secret:
        row = self.create(title, kind=kind, asked=why)
        Messages(self.record, actor=AGENT).create(f"Please fill in the secret {title}", brief=f"I need it {why}. Fill it in under Settings, Secrets, never in the chat.\n\nsecret {row.n}")
        return row

    @action
    def fill(self, n: int, field: str, value: str) -> Secret:
        if self.actor != USER:
            self._refuse("only you fill in a secret's value, under Settings, Secrets in the viewer")
        return self._stored(self.load(n), field, value)

    @action
    def store(self, n: int, field: str, path: str) -> Secret:
        made = Path(path)
        if not made.is_file():
            raise Refused(f"no file at {path}: write the value you made to a file, and the journal moves it into the secret")
        value = made.read_text().strip()
        made.unlink()
        return self._stored(self.load(n), field, value)

    @action(here=True)
    def run(self, name: str, *command: str, stdin: str = "", env: bool = False) -> str:
        row = self._named(name)
        self._shared_with_caller(row)
        checked_program(command, row.programs)
        fields = [SecretField.from_json(raw) for raw in row.secret_fields]
        values = ValuesFile(self.record.root).values()
        if row.is_waiting() or any(field.variable not in values for field in fields):
            raise Refused(f"secret {row.n}, {row.title}, has no value yet: ask for it with journal secret request, and the user fills it in")
        by_name = {field.name: values[field.variable] for field in fields}
        given = PLACEHOLDER.sub(lambda found: by_name.get(found[1], found[0]), stdin) if stdin else next(values[field.variable] for field in fields if field.hidden)
        masks = {values[field.variable]: row.title for field in fields if field.hidden}
        code = run_masked(command, {field.variable: values[field.variable] for field in fields} if env else {}, masks, f"{given}\n".encode())
        self.update(row.n, used=time.time())
        if code:
            raise SystemExit(code)
        return ""

    @action(here=True)
    def login(self, name: str, url: str) -> str:
        row = self._named(name)
        logins = BrowserLogins(self.record.root)
        saved = logins.record(row.title, url)
        merged = logins.merge()
        rewired = [provider().browser_logins(self.record.root.parent, merged) for provider in PROVIDERS.values()]
        self.update(row.n, session=time.time(), session_expires=logins.expires(saved))
        restart = " Restart the agent once so its browser tool reads the saved logins." if any(rewired) else ""
        return f"saved: the agent's own browser starts logged in to {url} from its next start.{restart}"

    @action
    def where(self) -> str:
        return str(ValuesFile(self.record.root).path)

    def _stored(self, row: Secret, field: str, value: str) -> Secret:
        if not value:
            raise Refused("a secret's value cannot be empty")
        ValuesFile(self.record.root).put(row.field(field).variable, value)
        return self.update(row.n, filled={**row.filled, field: time.time()}, asked="")

    def _named(self, name: str) -> Secret:
        if name.isdigit():
            return self.load(int(name))
        found = next((row for row in self.rows.every() if row.title.lower() == name.lower()), None)
        if found is None:
            raise Refused(f"no secret named {name!r}: journal secret all lists them")
        return found

    def _shared_with_caller(self, row: Secret) -> None:
        place = Environments(self.record, actor=SYSTEM).rows.by_title(self.record.env)
        helping = bool(place and place.data.get("kind") == "helper")
        if not row.helpers and (self.agent or helping):
            raise Refused(f"secret {row.n}, {row.title}, is for the main agent only; the user can share it with helpers and subagents under Settings, Secrets")

    def _purge(self) -> list[str]:
        gone = [row for row in self.rows.every(deleted=True) if row.deleted and time.time() - row.deleted > KEPT_DAYS * 86400]
        variables = [raw["variable"] for row in gone for raw in row.secret_fields]
        if variables:
            ValuesFile(self.record.root).drop(variables)
        return variables


resources_module.register(Secret)
types_module.register(Secrets)
