fix: OAuth2 token encoding and /auth/info route

src/envault/auth.py

@@ -16,6 +16,7 @@
 import time
 from typing import Any
 from urllib.error import URLError
+from urllib.parse import urlencode
 from urllib.request import Request, urlopen
 
 
@@ -164,7 +165,7 @@ def _introspect(self, token: str) -> AuthResult:
         import base64
 
         url = f"{self._provider_url}/introspect"
-        body = f"token={token}".encode()
+        body = urlencode({"token": token}).encode()
         headers: dict[str, str] = {
             "Content-Type": "application/x-www-form-urlencoded",
         }

src/envault/backup.py

@@ -92,16 +92,32 @@ def _get_backup_dir(project_dir: Path | str = ".") -> Path:
 
 
 def _load_manifest(backup_dir: Path) -> list[BackupEntry]:
-    """Load the backup manifest from disk."""
+    """Load the backup manifest from disk.
+
+    Skips individual corrupt entries rather than discarding the entire
+    manifest, preserving valid backups when one entry is malformed.
+    """
     manifest_path = backup_dir / BACKUP_MANIFEST
     if not manifest_path.exists():
         return []
     try:
         data = json.loads(manifest_path.read_text(encoding="utf-8"))
-        return [BackupEntry.from_dict(entry) for entry in data]
-    except (json.JSONDecodeError, KeyError):
+    except json.JSONDecodeError:
         return []
 
+    if not isinstance(data, list):
+        return []
+
+    entries: list[BackupEntry] = []
+    for entry in data:
+        if not isinstance(entry, dict):
+            continue
+        try:
+            entries.append(BackupEntry.from_dict(entry))
+        except (KeyError, TypeError):
+            continue
+    return entries
+
 
 def _save_manifest(backup_dir: Path, entries: list[BackupEntry]) -> None:
     """Save the backup manifest to disk."""

src/envault/cli.py

@@ -21,7 +21,7 @@
 from envault.diff import diff_env_files, format_diff
 from envault.encrypt import decrypt_env, encrypt_env
 from envault.history import format_history, get_env_history
-from envault.rotate import rotate_env_var
+from envault.rotate import rotate_env_file, rotate_env_var
 from envault.security_audit import (
     SecurityAuditResult,
     audit_env_file,
@@ -337,16 +337,12 @@ def rotate_all(
             console.print("Cancelled")
             raise typer.Exit(0)
 
-    rotated = 0
-    for key in sorted(vars.keys()):
-        success, new_val = rotate_env_var(
-            key,
-            env_file,
-            dry_run=dry_run,
-            audit=audit,
-        )
-        if success:
-            rotated += 1
+    new_values = rotate_env_file(
+        env_file,
+        dry_run=dry_run,
+        audit=audit,
+    )
+    rotated = len(new_values)
 
     if dry_run:
         console.print(f"[yellow]Dry run:[/yellow] Would rotate {rotated} variables in {env}")

src/envault/history.py

@@ -184,7 +184,9 @@ def _get_commits_for_file(file_path: Path, *, max_commits: int = 50) -> list[str
                 f"--max-count={max_commits}",
                 "--format=%H",
                 "--",
-                str(file_path),
+                # git pathspecs always use "/" — backslashes silently match
+                # nothing on Windows, which reads as "no history".
+                file_path.as_posix(),
             ],
             capture_output=True,
             text=True,
@@ -209,7 +211,7 @@ def _diff_at_commit(
     or changed.
     """
     # Get commit metadata
-    meta = _get_commit_meta(commit)
+    meta = _get_commit_meta(commit, cwd=file_path.parent)
     if meta is None:
         return []
 
@@ -274,18 +276,28 @@ def _diff_at_commit(
     return changes
 
 
-def _get_commit_meta(commit: str) -> dict | None:
-    """Get author, date, and message for a commit."""
+def _get_commit_meta(commit: str, cwd: Path | None = None) -> dict | None:
+    """Get author, date, and message for a commit.
+
+    Args:
+        commit: Commit hash (or ref) to describe.
+        cwd: Directory inside the repo that owns the commit. Without it the
+            lookup runs in the process CWD and silently returns None when the
+            caller's library code runs from a different repository.
+    """
     try:
+        # %x00 (NUL) separators: author names or subjects containing "|" would
+        # otherwise break the split and silently drop every change in the commit.
         result = subprocess.run(
-            ["git", "show", "-s", "--format=%an|%ai|%s", commit],
+            ["git", "show", "-s", "--format=%an%x00%ai%x00%s", commit],
             capture_output=True,
             text=True,
+            cwd=cwd if cwd is not None and Path(cwd).exists() else ".",
             timeout=10,
         )
         if result.returncode != 0:
             return None
-        parts = result.stdout.strip().split("|", 2)
+        parts = result.stdout.rstrip("\n").split("\x00", 2)
         if len(parts) != 3:
             return None
         return {"author": parts[0], "date": parts[1], "message": parts[2]}
@@ -344,7 +356,9 @@ def _get_relative_path(file_path: Path) -> str:
         if result.returncode == 0:
             repo_root = Path(result.stdout.strip())
             try:
-                return str(abs_path.relative_to(repo_root))
+                # POSIX separators: `git show <ref>:a\b.env` fails on Windows;
+                # git object paths always use "/".
+                return abs_path.relative_to(repo_root).as_posix()
             except ValueError:
                 pass
     except (subprocess.TimeoutExpired, FileNotFoundError):

src/envault/rotate.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import contextlib
 import os
+import re
 import secrets
 import string
 from pathlib import Path
@@ -176,3 +178,83 @@ def rotate_env_var(
         audit.log("rotate", key, env_file=str(env_file))
 
     return True, new_value
+
+
+def _atomic_write(path: Path, content: str) -> None:
+    """Write *content* to *path* atomically (temp file + os.replace).
+
+    A crash mid-write must never leave a truncated or half-rotated .env file.
+    """
+    tmp = path.with_name(f".{path.name}.rotate-tmp-{os.getpid()}")
+    try:
+        with open(tmp, "w") as f:
+            f.write(content)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(tmp, path)
+    except BaseException:
+        with contextlib.suppress(OSError):
+            os.unlink(tmp)
+        raise
+
+
+def rotate_env_file(
+    env_file: str | Path,
+    *,
+    length: int = 32,
+    exclude: set[str] | None = None,
+    dry_run: bool = False,
+    audit: AuditLogger | None = None,
+) -> dict[str, str]:
+    """Rotate every variable in a .env file with ONE atomic rewrite.
+
+    Args:
+        env_file: Path to the .env file.
+        length: Length of generated secrets.
+        exclude: Keys to leave untouched.
+        dry_run: If True, don't modify the file.
+        audit: Optional audit logger (one entry per rotated key).
+
+    Returns:
+        Mapping of key -> new value for every rotated key.
+    """
+    from dotenv import dotenv_values
+
+    env_file = Path(env_file)
+    exclude = exclude or set()
+    env_vars = dotenv_values(env_file)
+
+    plan: dict[str, str] = {}
+    for key, value in env_vars.items():
+        if key in exclude or value is None:
+            continue
+        plan[key] = rotate_value(key, value, length=length)
+
+    if dry_run or not plan:
+        return plan
+
+    lines = env_file.read_text().split("\n")
+    seen: set[str] = set()
+    out_lines: list[str] = []
+    key_line = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=")
+    for line in lines:
+        m = key_line.match(line)
+        if m and m.group(1) in plan and m.group(1) not in seen:
+            key = m.group(1)
+            seen.add(key)
+            new_value = plan[key]
+            if any(c in new_value for c in " #'\"\n\t"):
+                safe = new_value.replace("\\", "\\\\").replace('"', '\\"')
+                out_lines.append(f'{key}="{safe}"')
+            else:
+                out_lines.append(f"{key}={new_value}")
+        else:
+            out_lines.append(line)
+
+    _atomic_write(env_file, "\n".join(out_lines))
+
+    if audit:
+        for key in plan:
+            audit.log("rotate", key, env_file=str(env_file))
+
+    return plan

src/envault/serve.py

@@ -19,7 +19,6 @@
 import base64
 import json
 import os
-import secrets as _secrets
 import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
 from pathlib import Path
@@ -69,36 +68,6 @@ def _send_error(self, status: int, message: str) -> None:
         """Send a JSON error payload."""
         self._send_json({"error": message}, status=status)
 
-    def _check_auth(self) -> bool:
-        """Validate the Bearer token if API auth is enabled.
-
-        Returns True if the request is authorized (or auth is disabled).
-        Returns False if auth is required but missing/invalid (and sends 401).
-        """
-        if not self.api_key:
-            # Auth not configured — allow all requests
-            return True
-
-        auth_header = self.headers.get("Authorization", "")
-        if not auth_header:
-            self._send_error(401, "Unauthorized: valid Bearer token required")
-            return False
-
-        token = auth_header[len("Bearer ") :] if auth_header.startswith("Bearer ") else auth_header
-        if not token or not token.strip():
-            self._send_error(401, "Unauthorized: valid Bearer token required")
-            return False
-
-        if (
-            _secrets.compare_digest(token.strip(), self.api_key)
-            if self.api_key
-            else _secrets.compare_digest(token.strip(), "")
-        ):
-            return True
-
-        self._send_error(401, "Unauthorized: valid Bearer token required")
-        return False
-
     # ── Routing ──────────────────────────────────────────────────────────────
 
     def _check_bearer_token(self) -> bool:
@@ -330,6 +299,9 @@ def do_GET(self) -> None:  # noqa: N802 -- stdlib naming convention
         if path == "/health":
             # /health is always accessible (useful for load balancers)
             self._handle_health()
+        elif path == "/auth/info":
+            # /auth/info is always accessible so clients can discover auth methods
+            self._handle_auth_info()
         elif path == "/secrets":
             if not self._check_auth():
                 return