diff --git a/src/envault/auth.py b/src/envault/auth.py index 05bfd58..1155ea9 100644 --- a/src/envault/auth.py +++ b/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", } diff --git a/src/envault/backup.py b/src/envault/backup.py index 228caa3..3f83646 100644 --- a/src/envault/backup.py +++ b/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.""" diff --git a/src/envault/cli.py b/src/envault/cli.py index 8427297..2aa2357 100644 --- a/src/envault/cli.py +++ b/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}") diff --git a/src/envault/history.py b/src/envault/history.py index ed2622b..3ddea44 100644 --- a/src/envault/history.py +++ b/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 :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): diff --git a/src/envault/rotate.py b/src/envault/rotate.py index 18dad1b..9a55712 100644 --- a/src/envault/rotate.py +++ b/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 diff --git a/src/envault/serve.py b/src/envault/serve.py index d8ce597..fe47665 100644 --- a/src/envault/serve.py +++ b/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 diff --git a/src/envault/stores/__init__.py b/src/envault/stores/__init__.py index 7080641..1424649 100644 --- a/src/envault/stores/__init__.py +++ b/src/envault/stores/__init__.py @@ -14,6 +14,20 @@ class SecretStoreError(Exception): pass +def _is_missing_path_error(exc: BaseException) -> bool: + """Heuristically decide whether an exception means 'key does not exist'. + + Matches hvac's InvalidPath (checked by class name so hvac stays an optional + dependency) plus common HTTP-404 style messages. Anything else — auth + failures, connection errors, server errors — is NOT a missing key and must + surface to the caller instead of being silently swallowed. + """ + if type(exc).__name__ == "InvalidPath": + return True + message = str(exc).lower() + return "404" in message or "not found" in message or "path" in message and "missing" in message + + class SecretStore(ABC): """Abstract base class for secret store integrations.""" @@ -210,8 +224,10 @@ def get(self, key: str) -> str | None: ) data = response.get("data", {}).get("data", {}) return data.get("value") - except Exception: - return None + except Exception as exc: + if _is_missing_path_error(exc): + return None + raise SecretStoreError(f"Vault read failed for {key!r}: {exc}") from exc def set(self, key: str, value: str) -> bool: client = self._get_client() @@ -230,8 +246,10 @@ def delete(self, key: str) -> bool: mount_point=self.mount_point, ) return True - except Exception: - return False + except Exception as exc: + if _is_missing_path_error(exc): + return False + raise SecretStoreError(f"Vault delete failed for {key!r}: {exc}") from exc def list_keys(self, prefix: str = "") -> list[str]: client = self._get_client() @@ -242,8 +260,10 @@ def list_keys(self, prefix: str = "") -> list[str]: mount_point=self.mount_point, ) return response.get("data", {}).get("keys", []) - except Exception: - return [] + except Exception as exc: + if _is_missing_path_error(exc): + return [] + raise SecretStoreError(f"Vault list failed at {list_path!r}: {exc}") from exc class DopplerStore(SecretStore): @@ -279,6 +299,26 @@ def get(self, key: str) -> str | None: return secrets[key].get("raw", "").strip() or secrets[key].get("computed", "").strip() return None + def get_many(self, keys: list[str]) -> dict[str, str]: + """Batch fetch: Doppler returns the whole config's secrets per request, + so one request serves all keys instead of one request per key.""" + import requests + + url = f"{self._base_url}/configs/config/secrets" + params = {"project": self.project, "config": self.config} + resp = requests.get(url, headers=self._headers(), params=params, timeout=10) + if resp.status_code != 200: + return {} + data = resp.json() + secrets = data.get("secrets", {}) + result: dict[str, str] = {} + for key in keys: + if key in secrets: + value = secrets[key].get("raw", "").strip() or secrets[key].get("computed", "").strip() + if value: + result[key] = value + return result + def set(self, key: str, value: str) -> bool: import requests @@ -346,7 +386,10 @@ def _api_post(self, path: str, data: dict) -> bool: return resp.status_code in (200, 201) def get(self, key: str) -> str | None: - items = self._api_get(f"/v1/vaults/{self.vault_id}/items?filter=title%20eq%20%22{key}%22") + from urllib.parse import quote + + encoded_key = quote(key, safe="") + items = self._api_get(f"/v1/vaults/{self.vault_id}/items?filter=title%20eq%20%22{encoded_key}%22") if not items: return None item_list = items if isinstance(items, list) else items.get("items", []) @@ -370,9 +413,12 @@ def set(self, key: str, value: str) -> bool: return self._api_post(f"/v1/vaults/{self.vault_id}/items", payload) def delete(self, key: str) -> bool: + from urllib.parse import quote + import requests - items = self._api_get(f"/v1/vaults/{self.vault_id}/items?filter=title%20eq%20%22{key}%22") + encoded_key = quote(key, safe="") + items = self._api_get(f"/v1/vaults/{self.vault_id}/items?filter=title%20eq%20%22{encoded_key}%22") if not items: return False item_list = items if isinstance(items, list) else items.get("items", []) diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..432f74d --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import json + +from envault.auth import OAuth2Auth + + +def test_oauth2_introspection_url_encodes_reserved_token_characters(monkeypatch): + captured: dict[str, object] = {} + + class _Response: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + return False + + def read(self): + return json.dumps({"active": True, "sub": "synthetic-user"}).encode() + + def fake_urlopen(request, timeout): + captured["request"] = request + captured["timeout"] = timeout + return _Response() + + monkeypatch.setattr("envault.auth.urlopen", fake_urlopen) + + result = OAuth2Auth(provider_url="https://identity.example", strategy="introspect").check( + {"Authorization": "Bearer token+with&reserved=value"} + ) + + assert result.success + request = captured["request"] + assert request.data == b"token=token%2Bwith%26reserved%3Dvalue" + assert captured["timeout"] == 10 diff --git a/tests/test_auth_coverage.py b/tests/test_auth_coverage.py new file mode 100644 index 0000000..e233ceb --- /dev/null +++ b/tests/test_auth_coverage.py @@ -0,0 +1,457 @@ +"""Coverage-driven regression tests for envault.auth module. + +Closes gaps in BearerAuth, ApiKeyAuth, OAuth2Auth (userinfo strategy, +cache expiry, scope/audience validation, error paths), MultiAuth fallback +logic, and build_auth_from_env factory. +""" +from __future__ import annotations + +import json +import time + +from envault.auth import ( + ApiKeyAuth, + AuthResult, + BearerAuth, + MultiAuth, + OAuth2Auth, + build_auth_from_env, +) + +# ── AuthResult ─────────────────────────────────────────────────────────── + + +class TestAuthResult: + def test_ok_default_identity(self): + r = AuthResult.ok() + assert r.success is True + assert r.identity == "anonymous" + assert r.error_status == 401 + assert r.error_message == "" + + def test_ok_custom_identity(self): + r = AuthResult.ok(identity="user:alice") + assert r.success is True + assert r.identity == "user:alice" + + def test_fail_default(self): + r = AuthResult.fail() + assert r.success is False + assert r.error_status == 401 + assert r.error_message == "Unauthorized" + + def test_fail_custom(self): + r = AuthResult.fail(status=403, message="Forbidden") + assert r.success is False + assert r.error_status == 403 + assert r.error_message == "Forbidden" + + +# ── BearerAuth ─────────────────────────────────────────────────────────── + + +class TestBearerAuth: + def test_valid_token(self): + auth = BearerAuth("secret-token-12345") + result = auth.check({"Authorization": "Bearer secret-token-12345"}) + assert result.success is True + assert "bearer:" in result.identity + + def test_missing_header(self): + auth = BearerAuth("token") + result = auth.check({}) + assert result.success is False + assert result.error_status == 401 + + def test_non_bearer_scheme(self): + auth = BearerAuth("token") + result = auth.check({"Authorization": "Basic dXNlcjpwYXNz"}) + assert result.success is False + assert result.error_status == 401 + assert "Bearer token required" in result.error_message + + def test_wrong_token(self): + auth = BearerAuth("correct") + result = auth.check({"Authorization": "Bearer wrong"}) + assert result.success is False + assert result.error_status == 403 + assert "invalid token" in result.error_message + + def test_empty_bearer_value(self): + auth = BearerAuth("token") + result = auth.check({"Authorization": "Bearer "}) + assert result.success is False + assert result.error_status == 403 + + +# ── ApiKeyAuth ─────────────────────────────────────────────────────────── + + +class TestApiKeyAuth: + def test_valid_key_single(self): + auth = ApiKeyAuth("my-api-key") + result = auth.check({"X-Api-Key": "my-api-key"}) + assert result.success is True + assert "api_key:" in result.identity + + def test_valid_key_uppercase_header(self): + auth = ApiKeyAuth("my-api-key") + result = auth.check({"X-API-KEY": "my-api-key"}) + assert result.success is True + + def test_comma_separated_keys(self): + auth = ApiKeyAuth("key1, key2, key3") + assert auth.check({"X-Api-Key": "key1"}).success + assert auth.check({"X-Api-Key": "key2"}).success + assert auth.check({"X-Api-Key": "key3"}).success + assert not auth.check({"X-Api-Key": "key4"}).success + + def test_list_keys(self): + auth = ApiKeyAuth(["a", "b"]) + assert auth.check({"X-Api-Key": "a"}).success + assert not auth.check({"X-Api-Key": "c"}).success + + def test_missing_key(self): + auth = ApiKeyAuth("key") + result = auth.check({}) + assert result.success is False + assert result.error_status == 401 + assert "API key required" in result.error_message + + def test_invalid_key(self): + auth = ApiKeyAuth("valid") + result = auth.check({"X-Api-Key": "invalid"}) + assert result.success is False + assert result.error_status == 403 + assert "invalid API key" in result.error_message + + def test_empty_string_keys_stripped(self): + auth = ApiKeyAuth("key1,, ,key2") + assert len(auth._keys) == 2 + + +# ── OAuth2Auth ─────────────────────────────────────────────────────────── + + +def _make_response(status: int, body: dict): + """Create a mock urlopen response context manager.""" + + class _Resp: + def __init__(self): + self.status = status + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + def read(self): + return json.dumps(body).encode() + + return _Resp() + + +class TestOAuth2AuthUserinfo: + def test_userinfo_success(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"sub": "user1", "email": "u@test.com"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({"Authorization": "Bearer valid-token"}) + assert result.success is True + assert "oauth2:user1" in result.identity + + def test_userinfo_uses_email_fallback(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"email": "fallback@test.com"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + assert "fallback@test.com" in result.identity + + def test_userinfo_rejected(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(401, {"error": "invalid_token"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({"Authorization": "Bearer bad-token"}) + assert result.success is False + assert result.error_status == 401 + + def test_missing_bearer_prefix(self): + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({"Authorization": "Basic abc"}) + assert result.success is False + assert result.error_status == 401 + assert "Bearer token required" in result.error_message + + def test_no_auth_header(self): + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({}) + assert result.success is False + assert result.error_status == 401 + + +class TestOAuth2AuthIntrospect: + def test_introspect_active(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"active": True, "sub": "client1"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", strategy="introspect") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + + def test_introspect_inactive(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"active": False}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", strategy="introspect") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is False + assert result.error_status == 401 + assert "not active" in result.error_message + + def test_introspect_with_client_creds(self, monkeypatch): + captured = {} + + def fake_urlopen(req, timeout=10): + captured["headers"] = dict(req.headers) + return _make_response(200, {"active": True, "sub": "svc"}) + + monkeypatch.setattr("envault.auth.urlopen", fake_urlopen) + auth = OAuth2Auth( + provider_url="https://idp.example", + strategy="introspect", + client_id="cid", + client_secret="csec", + ) + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + assert "Basic" in captured["headers"].get("Authorization", "") + + +class TestOAuth2AuthCache: + def test_cache_hit(self, monkeypatch): + call_count = 0 + + def fake_urlopen(req, timeout=10): + nonlocal call_count + call_count += 1 + return _make_response(200, {"sub": "cached-user"}) + + monkeypatch.setattr("envault.auth.urlopen", fake_urlopen) + auth = OAuth2Auth(provider_url="https://idp.example", cache_ttl=60) + + r1 = auth.check({"Authorization": "Bearer tok"}) + r2 = auth.check({"Authorization": "Bearer tok"}) + assert r1.success and r2.success + assert call_count == 1 # Second call served from cache + + def test_cache_expiry(self, monkeypatch): + call_count = 0 + + def fake_urlopen(req, timeout=10): + nonlocal call_count + call_count += 1 + return _make_response(200, {"sub": "user"}) + + monkeypatch.setattr("envault.auth.urlopen", fake_urlopen) + auth = OAuth2Auth(provider_url="https://idp.example", cache_ttl=1) + + auth.check({"Authorization": "Bearer tok"}) + assert call_count == 1 + + # Simulate cache expiry by manipulating internal state + for token_key in list(auth._cache.keys()): + identity, _ = auth._cache[token_key] + auth._cache[token_key] = (identity, time.monotonic() - 1) + + auth.check({"Authorization": "Bearer tok"}) + assert call_count == 2 # Cache expired, re-validated + + +class TestOAuth2AuthScopeAudience: + def test_scope_present(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"sub": "u", "scope": "read write admin"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", required_scope="read write") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + + def test_scope_missing(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"sub": "u", "scope": "read"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", required_scope="read write") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is False + assert result.error_status == 403 + assert "missing scope" in result.error_message + assert "write" in result.error_message + + def test_audience_string_match(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"sub": "u", "aud": "my-api"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", required_audience="my-api") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + + def test_audience_list_match(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"sub": "u", "aud": ["api-a", "api-b"]}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", required_audience="api-b") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + + def test_audience_mismatch(self, monkeypatch): + monkeypatch.setattr( + "envault.auth.urlopen", + lambda req, timeout=10: _make_response(200, {"sub": "u", "aud": "other-api"}), + ) + auth = OAuth2Auth(provider_url="https://idp.example", required_audience="my-api") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is False + assert result.error_status == 403 + assert "invalid audience" in result.error_message + + +class TestOAuth2AuthErrors: + def test_url_error(self, monkeypatch): + from urllib.error import URLError + + def fake_urlopen(req, timeout=10): + raise URLError(reason="connection refused") + + monkeypatch.setattr("envault.auth.urlopen", fake_urlopen) + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is False + assert result.error_status == 502 + assert "unreachable" in result.error_message + + def test_generic_exception(self, monkeypatch): + def fake_urlopen(req, timeout=10): + raise RuntimeError("unexpected") + + monkeypatch.setattr("envault.auth.urlopen", fake_urlopen) + auth = OAuth2Auth(provider_url="https://idp.example") + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is False + assert result.error_status == 502 + assert "validation error" in result.error_message + + +# ── MultiAuth ──────────────────────────────────────────────────────────── + + +class TestMultiAuth: + def test_open_mode_no_backends(self): + auth = MultiAuth() + result = auth.check({}) + assert result.success is True + assert result.identity == "open" + assert auth.is_enabled is False + + def test_empty_list_is_open(self): + auth = MultiAuth([]) + assert auth.is_enabled is False + assert auth.check({}).success is True + + def test_first_backend_wins(self): + auth = MultiAuth([BearerAuth("tok"), ApiKeyAuth("key")]) + assert auth.is_enabled is True + result = auth.check({"Authorization": "Bearer tok"}) + assert result.success is True + assert "bearer:" in result.identity + + def test_falls_through_to_second(self): + auth = MultiAuth([BearerAuth("tok"), ApiKeyAuth("key")]) + result = auth.check({"X-Api-Key": "key"}) + assert result.success is True + assert "api_key:" in result.identity + + def test_returns_most_specific_failure(self): + """When all backends fail, prefer 403 (wrong creds) over 401 (no creds).""" + auth = MultiAuth([BearerAuth("correct"), ApiKeyAuth("correct")]) + # Wrong bearer → 403; missing api key → 401. Should return 403. + result = auth.check({"Authorization": "Bearer wrong"}) + assert result.success is False + assert result.error_status == 403 + + def test_all_missing_returns_401(self): + auth = MultiAuth([BearerAuth("tok"), ApiKeyAuth("key")]) + result = auth.check({}) + assert result.success is False + assert result.error_status == 401 + + +# ── build_auth_from_env ────────────────────────────────────────────────── + + +class TestBuildAuthFromEnv: + def test_empty_env_is_open(self, monkeypatch): + for var in [ + "ENVAULT_API_TOKEN", + "ENVAULT_API_KEY", + "ENVAULT_OAUTH2_URL", + ]: + monkeypatch.delenv(var, raising=False) + auth = build_auth_from_env() + assert auth.is_enabled is False + + def test_bearer_only(self, monkeypatch): + monkeypatch.setenv("ENVAULT_API_TOKEN", "my-token") + monkeypatch.delenv("ENVAULT_API_KEY", raising=False) + monkeypatch.delenv("ENVAULT_OAUTH2_URL", raising=False) + auth = build_auth_from_env() + assert auth.is_enabled is True + result = auth.check({"Authorization": "Bearer my-token"}) + assert result.success is True + + def test_api_key_only(self, monkeypatch): + monkeypatch.delenv("ENVAULT_API_TOKEN", raising=False) + monkeypatch.setenv("ENVAULT_API_KEY", "k1,k2") + monkeypatch.delenv("ENVAULT_OAUTH2_URL", raising=False) + auth = build_auth_from_env() + assert auth.is_enabled is True + assert auth.check({"X-Api-Key": "k1"}).success + assert auth.check({"X-Api-Key": "k2"}).success + + def test_oauth2_configured(self, monkeypatch): + monkeypatch.delenv("ENVAULT_API_TOKEN", raising=False) + monkeypatch.delenv("ENVAULT_API_KEY", raising=False) + monkeypatch.setenv("ENVAULT_OAUTH2_URL", "https://idp.example") + monkeypatch.setenv("ENVAULT_OAUTH2_STRATEGY", "introspect") + monkeypatch.setenv("ENVAULT_OAUTH2_CLIENT_ID", "cid") + monkeypatch.setenv("ENVAULT_OAUTH2_CLIENT_SECRET", "csec") + monkeypatch.setenv("ENVAULT_OAUTH2_SCOPE", "read") + monkeypatch.setenv("ENVAULT_OAUTH2_AUDIENCE", "api") + auth = build_auth_from_env() + assert auth.is_enabled is True + assert len(auth._backends) == 1 + backend = auth._backends[0] + assert isinstance(backend, OAuth2Auth) + assert backend._strategy == "introspect" + assert backend._required_scope == "read" + assert backend._required_audience == "api" + + def test_all_backends_combined(self, monkeypatch): + monkeypatch.setenv("ENVAULT_API_TOKEN", "tok") + monkeypatch.setenv("ENVAULT_API_KEY", "key") + monkeypatch.setenv("ENVAULT_OAUTH2_URL", "https://idp.example") + auth = build_auth_from_env() + assert len(auth._backends) == 3 diff --git a/tests/test_backup_manifest.py b/tests/test_backup_manifest.py new file mode 100644 index 0000000..24302c9 --- /dev/null +++ b/tests/test_backup_manifest.py @@ -0,0 +1,80 @@ +"""Tests for backup manifest loading resilience.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from envault.backup import BACKUP_MANIFEST, _load_manifest + + +def _write_manifest(backup_dir: Path, data: list[dict]) -> None: + """Helper to write raw manifest JSON.""" + manifest_path = backup_dir / BACKUP_MANIFEST + manifest_path.write_text(json.dumps(data), encoding="utf-8") + + +def test_load_manifest_skips_corrupt_entries(tmp_path: Path) -> None: + """A single corrupt entry must not discard valid entries. + + Regression: previously a KeyError on any entry caused the entire + manifest to be silently discarded, losing all valid backups. + """ + valid_entry = { + "name": "good-backup", + "source_file": ".env", + "backup_path": str(tmp_path / "good-backup"), + "timestamp": "2026-08-10T00:00:00+00:00", + "encrypted": False, + } + corrupt_entry = {"name": "missing-fields"} # missing source_file, backup_path, timestamp + + _write_manifest(tmp_path, [valid_entry, corrupt_entry]) + + entries = _load_manifest(tmp_path) + + assert len(entries) == 1 + assert entries[0].name == "good-backup" + assert entries[0].source_file == ".env" + + +def test_load_manifest_all_corrupt_returns_empty(tmp_path: Path) -> None: + """When every entry is corrupt, return empty list without raising.""" + _write_manifest(tmp_path, [{"bad": True}, {"also_bad": True}]) + + entries = _load_manifest(tmp_path) + + assert entries == [] + + +def test_load_manifest_valid_json_but_not_list(tmp_path: Path) -> None: + """A manifest that is valid JSON but not a list returns empty.""" + manifest_path = tmp_path / BACKUP_MANIFEST + manifest_path.write_text('{"not": "a list"}', encoding="utf-8") + + entries = _load_manifest(tmp_path) + + assert entries == [] + + +def test_load_manifest_preserves_order(tmp_path: Path) -> None: + """Valid entries are returned in their original order.""" + entries_data = [ + { + "name": f"backup-{i}", + "source_file": f".env.{i}", + "backup_path": str(tmp_path / f"backup-{i}"), + "timestamp": f"2026-08-10T00:0{i}:00+00:00", + "encrypted": False, + } + for i in range(5) + ] + # Insert a corrupt entry in the middle + entries_data.insert(2, {"corrupt": True}) + + _write_manifest(tmp_path, entries_data) + + result = _load_manifest(tmp_path) + + assert len(result) == 5 + assert [e.name for e in result] == [f"backup-{i}" for i in range(5)] diff --git a/tests/test_cli_edge_cases.py b/tests/test_cli_edge_cases.py index dcd8c88..6c2452e 100644 --- a/tests/test_cli_edge_cases.py +++ b/tests/test_cli_edge_cases.py @@ -31,9 +31,7 @@ def _make_config(tmp_path, env_map): """Create minimal .envault.yml with list-formatted environments.""" config = { "project": "test", - "environments": [ - {"name": name, "env_file": path} for name, path in env_map.items() - ], + "environments": [{"name": name, "env_file": path} for name, path in env_map.items()], } config_path = tmp_path / ".envault.yml" with open(config_path, "w") as f: @@ -171,9 +169,7 @@ def test_package_data_includes_py_typed(self): with open(pyproject, "rb") as f: data = tomllib.load(f) pkg_data = data.get("tool", {}).get("setuptools", {}).get("package-data", {}) - assert "envault" in pkg_data, ( - "Expected [tool.setuptools.package-data] section for 'envault'" - ) + assert "envault" in pkg_data, "Expected [tool.setuptools.package-data] section for 'envault'" assert "py.typed" in pkg_data["envault"], ( f"Expected 'py.typed' in package-data for envault, got {pkg_data['envault']}" ) @@ -184,8 +180,6 @@ def test_ruff_known_first_party(self): pyproject = Path(__file__).parent.parent / "pyproject.toml" with open(pyproject, "rb") as f: data = tomllib.load(f) - isort_cfg = ( - data.get("tool", {}).get("ruff", {}).get("lint", {}).get("isort", {}) - ) + isort_cfg = data.get("tool", {}).get("ruff", {}).get("lint", {}).get("isort", {}) kfp = isort_cfg.get("known-first-party", []) assert kfp == ["envault"], f"known-first-party should be ['envault'], got {kfp}" diff --git a/tests/test_history.py b/tests/test_history.py index 9807e08..4cd0ebb 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -1,128 +1,153 @@ -"""Auto-generated tests for envault.history.""" +"""Tests for git-based .env change history (src/envault/history.py).""" from __future__ import annotations -from pathlib import Path +import shutil +import subprocess +from types import SimpleNamespace import pytest -# ── Fixtures ──────────────────────────────────────────────────────────────── +from envault.history import ( + EnvFileHistory, + _get_commit_meta, + _mask_value, + _parse_env_content, + get_env_history, +) +GIT_AVAILABLE = shutil.which("git") is not None -# ── get_env_history ────────────────────────────────────────────────────────────── +# ── _parse_env_content ──────────────────────────────────────────────────── -def test_get_env_history(file_path=...): - """Get the change history of an .env file from git.""" - # TODO: implement test for get_env_history - # result = history.get_env_history(...) - # assert result is not None +def test_parse_env_content_basic(): + content = "A=1\nB = two\n\n# comment\nNO_EQUALS_LINE\n" + parsed = _parse_env_content(content) + assert parsed == {"A": "1", "B": "two"} -def test_get_env_history_file_path_edge_cases(tmp_path): - """Edge cases for get_env_history param file_path.""" - from envault.history import get_env_history - # Nonexistent path — should handle gracefully - result = get_env_history(Path("/no/such/path")) - assert result is not None or result is None # smoke test - # Tmp path — no git history - result2 = get_env_history(tmp_path / ".env") - assert result2 is not None or result2 is None # smoke test +def test_parse_env_content_strips_symmetric_quotes(): + content = 'A="quoted"\nB=' + "'single'\n" + "C=un\"matched\n" + parsed = _parse_env_content(content) + assert parsed["A"] == "quoted" + assert parsed["B"] == "single" + assert parsed["C"] == 'un"matched' -# ── get_env_history_multiple ────────────────────────────────────────────────────────────── +# ── _get_commit_meta ────────────────────────────────────────────────────── -def test_get_env_history_multiple(file_paths=...): - """Get change history for multiple .env files.""" - # TODO: implement test for get_env_history_multiple - # result = history.get_env_history_multiple(...) - # assert result is not None +def test_commit_meta_survives_pipe_in_author(monkeypatch): + """An author name containing '|' must not silently drop the metadata.""" + payload = "Jane | Doe\x002026-01-02 03:04:05 +0000\x00rotate staging keys\n" + def fake_run(*args, **kwargs): + return SimpleNamespace(returncode=0, stdout=payload) + + monkeypatch.setattr(subprocess, "run", fake_run) + meta = _get_commit_meta("abc123") + assert meta == { + "author": "Jane | Doe", + "date": "2026-01-02 03:04:05 +0000", + "message": "rotate staging keys", + } + + +def test_commit_meta_subject_with_pipes(monkeypatch): + """Only the FIRST two separators split; the subject keeps any '|'. + + The legacy '|' delimiter truncated subjects at the first pipe. + """ + payload = "auth\x00date\x00fix: a | b | c\n" + + def fake_run(*args, **kwargs): + return SimpleNamespace(returncode=0, stdout=payload) + + monkeypatch.setattr(subprocess, "run", fake_run) + meta = _get_commit_meta("abc123") + assert meta is not None + assert meta["message"] == "fix: a | b | c" + + +def test_commit_meta_git_failure_returns_none(monkeypatch): + def fake_run(*args, **kwargs): + return SimpleNamespace(returncode=1, stdout="") + + monkeypatch.setattr(subprocess, "run", fake_run) + assert _get_commit_meta("deadbeef") is None + + +# ── _mask_value ─────────────────────────────────────────────────────────── + + +def test_mask_value_long_secret_is_masked(): + masked = _mask_value("supersecretvalue_0123456789") + assert masked.startswith("supersec") + assert "..." in masked + assert "0123456789" not in masked + + +def test_mask_value_short_or_pathlike_kept(): + assert _mask_value("short") == "short" + assert _mask_value("/usr/local/bin") == "/usr/local/bin" + + +# ── end-to-end against a real temp repo ─────────────────────────────────── -@pytest.mark.parametrize( - "file_paths", - [ - pytest.param("", id="empty_string"), - pytest.param(" ", id="whitespace"), - pytest.param( - "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", - id="long_string", - ), - pytest.param("héllo wörld", id="unicode"), - pytest.param("line1\nline2", id="with_newline"), - ], -) -def test_get_env_history_multiple_file_paths_edge_cases(file_paths): - """Edge cases for get_env_history_multiple param file_paths.""" - # TODO: call history.get_env_history_multiple with edge-case file_paths - pass - - -# ── format_history ────────────────────────────────────────────────────────────── - - -def test_format_history(history=...): - """Format an EnvFileHistory as a human-readable string.""" - # TODO: implement test for format_history - # result = history.format_history(...) - # assert result is not None - - -@pytest.mark.parametrize( - "history", - [ - pytest.param("", id="empty_string"), - pytest.param(" ", id="whitespace"), - pytest.param( - "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", - id="long_string", - ), - pytest.param("héllo wörld", id="unicode"), - pytest.param("line1\nline2", id="with_newline"), - ], -) -def test_format_history_history_edge_cases(history): - """Edge cases for format_history param history.""" - # TODO: call history.format_history with edge-case history - pass +def _git(cwd, *args): + subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True) -# ── EnvChange ─────────────────────────────────────────────────────────────── +@pytest.mark.skipif(not GIT_AVAILABLE, reason="git not available") +def test_get_env_history_detects_changes_in_nested_file(tmp_path): + """Regression: nested paths must be passed to git with '/' separators. -class TestEnvChange: - """Tests for EnvChange.""" + Windows backslash pathspecs matched nothing, so history came back empty + (silent failure) for every .env file not at the repo root. + """ + subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) + _git(tmp_path, "config", "user.email", "test@example.com") + _git(tmp_path, "config", "user.name", "Test User") - pass + nested = tmp_path / "config" + nested.mkdir() + env_file = nested / ".env" + env_file.write_text("A=1\nB=keep\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-m", "initial") + env_file.write_text("A=2\nB=keep\nC=new\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-m", "update values") -# ── EnvFileHistory ─────────────────────────────────────────────────────────────── + history = get_env_history(env_file) + assert isinstance(history, EnvFileHistory) + actions = {(c.key, c.action) for c in history.changes} + assert ("A", "changed") in actions + assert ("C", "added") in actions + # Unchanged key may only be recorded once, as the initial-commit add. + b_actions = [c.action for c in history.changes if c.key == "B"] + assert b_actions == ["added"] -class TestEnvFileHistory: - """Tests for EnvFileHistory.""" +@pytest.mark.skipif(not GIT_AVAILABLE, reason="git not available") +def test_get_env_history_key_filter(tmp_path): + subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) + _git(tmp_path, "config", "user.email", "test@example.com") + _git(tmp_path, "config", "user.name", "Test User") - def test_total_changes( - self, - ): - """Smoke test for EnvFileHistory.total_changes.""" - # TODO: implement test for EnvFileHistory.total_changes - # obj = EnvFileHistory(...) - # result = obj.total_changes(...) - # assert result is not None + env_file = tmp_path / ".env" + env_file.write_text("A=1\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-m", "initial") - def test_to_dict(self, mask_values=True): - """Serialize history as a dict suitable for JSON output.""" - # TODO: implement test for EnvFileHistory.to_dict - # obj = EnvFileHistory(...) - # result = obj.to_dict(...) - # assert result is not None + env_file.write_text("A=2\n", encoding="utf-8") + _git(tmp_path, "add", ".") + _git(tmp_path, "commit", "-m", "change a") - def test_to_json(self, mask_values=True, indent=2): - """Serialize history as a JSON string.""" - # TODO: implement test for EnvFileHistory.to_json - # obj = EnvFileHistory(...) - # result = obj.to_json(...) - # assert result is not None + history = get_env_history(env_file, key_filter="A") + assert history.total_changes >= 1 + assert all(c.key == "A" for c in history.changes) diff --git a/tests/test_rotate_all_atomic.py b/tests/test_rotate_all_atomic.py new file mode 100644 index 0000000..67aeb9c --- /dev/null +++ b/tests/test_rotate_all_atomic.py @@ -0,0 +1,67 @@ +"""Tests for atomic single-pass rotation (rotate_env_file).""" + +from pathlib import Path + +from dotenv import dotenv_values + +from envault.rotate import rotate_env_file + + +def _write(tmp_path: Path, content: str) -> Path: + p = tmp_path / ".env" + p.write_text(content) + return p + + +def test_rotate_env_file_rotates_all_keys(tmp_path): + env = _write(tmp_path, "DB_PASSWORD=old\nAPI_KEY=old\nNOTE=keepme\n") + plan = rotate_env_file(env, exclude={"NOTE"}) + assert set(plan) == {"DB_PASSWORD", "API_KEY"} + new = dotenv_values(env) + assert new["DB_PASSWORD"] != "old" and new["DB_PASSWORD"] + assert new["API_KEY"] != "old" and new["API_KEY"] + assert new["NOTE"] == "keepme" + + +def test_rotate_env_file_dry_run_leaves_file_untouched(tmp_path): + original = "DB_PASSWORD=old\nAPI_KEY=old\n" + env = _write(tmp_path, original) + plan = rotate_env_file(env, dry_run=True) + assert set(plan) == {"DB_PASSWORD", "API_KEY"} + assert env.read_text() == original + + +def test_rotate_env_file_exclude(tmp_path): + env = _write(tmp_path, "DB_PASSWORD=old\nKEEP_ME=untouched\n") + rotate_env_file(env, exclude={"KEEP_ME"}) + assert dotenv_values(env)["KEEP_ME"] == "untouched" + + +def test_rotate_env_file_single_rewrite_is_atomic(tmp_path): + """No temp files left behind; content replaced via os.replace.""" + env = _write(tmp_path, "A=1\nB=2\nC=3\n") + rotate_env_file(env) + leftovers = list(tmp_path.glob(".*rotate-tmp-*")) + assert leftovers == [] + # comments/blank lines preserved + text = env.read_text() + assert all(line.startswith(k + "=") for k, line in zip("ABC", text.splitlines(), strict=True)) + + +def test_rotate_env_file_preserves_comments_and_order(tmp_path): + env = _write(tmp_path, "# creds\nA=1\n\nB=2 # trailing comment\n") + rotate_env_file(env) + lines = env.read_text().splitlines() + assert lines[0] == "# creds" + assert lines[2] == "" + assert lines[1].startswith("A=") and lines[3].startswith("B=") + + +def test_rotate_env_file_audit_entries(tmp_path): + from envault.audit import AuditLogger + + log = tmp_path / "audit.log" + env = _write(tmp_path, "A=1\nB=2\n") + rotate_env_file(env, audit=AuditLogger(log)) + entries = [line for line in log.read_text().splitlines() if line] + assert len(entries) == 2 diff --git a/tests/test_serve.py b/tests/test_serve.py index d9d8317..452d487 100644 --- a/tests/test_serve.py +++ b/tests/test_serve.py @@ -675,6 +675,31 @@ def test_secrets_trailing_slash(self): assert handler._sent_status == 200 assert "keys" in handler._sent_json + def test_auth_info_endpoint_accessible(self): + """GET /auth/info should return auth configuration without requiring auth.""" + store = _FakeStore({}) + handler = _make_handler(store, api_key="secret-token") + handler.path = "/auth/info" + handler.do_GET() + + assert handler._sent_status == 200 + data = handler._sent_json + assert "auth_mode" in data + assert data["auth_mode"] == "bearer" + assert data["requires_auth"] is True + + def test_auth_info_no_auth_configured(self): + """GET /auth/info should show 'any' mode when no api_key is set.""" + store = _FakeStore({}) + handler = _make_handler(store, api_key=None) + handler.path = "/auth/info" + handler.do_GET() + + assert handler._sent_status == 200 + data = handler._sent_json + assert data["auth_mode"] == "any" + assert data["requires_auth"] is False + # ── Tests: API Authentication ────────────────────────────────────────────────── diff --git a/tests/test_stores_integration.py b/tests/test_stores_integration.py index e18c000..95cebdd 100644 --- a/tests/test_stores_integration.py +++ b/tests/test_stores_integration.py @@ -241,12 +241,61 @@ def test_list_keys_empty(self): store = VaultStore(token="s.test") mock_client = MagicMock() - mock_client.secrets.kv.v2.list_secrets.side_effect = Exception("no keys") + mock_client.secrets.kv.v2.list_secrets.side_effect = Exception("404 path not found") with patch.object(store, "_get_client", return_value=mock_client): keys = store.list_keys() assert keys == [] + def test_get_raises_on_real_error(self): + """Auth/connection/server errors must NOT be reported as 'missing key'.""" + from envault.stores import SecretStoreError, VaultStore + + store = VaultStore(token="s.test") + mock_client = MagicMock() + mock_client.secrets.kv.v2.read_secret.side_effect = Exception("permission denied") + + with patch.object(store, "_get_client", return_value=mock_client), pytest.raises( + SecretStoreError, match="Vault read failed" + ): + store.get("MY_KEY") + + def test_delete_raises_on_real_error(self): + from envault.stores import SecretStoreError, VaultStore + + store = VaultStore(token="s.test") + mock_client = MagicMock() + mock_client.secrets.kv.v2.delete_metadata_and_all_versions.side_effect = Exception( + "connection refused" + ) + + with patch.object(store, "_get_client", return_value=mock_client), pytest.raises( + SecretStoreError, match="Vault delete failed" + ): + store.delete("OLD_KEY") + + def test_list_raises_on_real_error(self): + from envault.stores import SecretStoreError, VaultStore + + store = VaultStore(token="s.test") + mock_client = MagicMock() + mock_client.secrets.kv.v2.list_secrets.side_effect = Exception("500 internal server error") + + with patch.object(store, "_get_client", return_value=mock_client), pytest.raises( + SecretStoreError, match="Vault list failed" + ): + store.list_keys() + + def test_hvac_invalid_path_class_is_missing(self): + """hvac.exceptions.InvalidPath (matched by class name) means 'not found'.""" + from envault.stores import _is_missing_path_error + + class InvalidPath(Exception): + pass + + assert _is_missing_path_error(InvalidPath("missing")) is True + assert _is_missing_path_error(Exception("connection refused")) is False + def test_vault_auth_fails(self): from envault.stores import SecretStoreError, VaultStore @@ -310,6 +359,32 @@ def test_get_falls_back_to_computed(self): result = store.get("MY_KEY") assert result == "computed_val" + def test_get_many_single_request(self): + import responses + + from envault.stores import DopplerStore + + store = DopplerStore(project="myapp", config="prd", token="dp-test") + url = "https://api.doppler.com/v3/configs/config/secrets" + + with responses.RequestsMock() as rsps: + rsps.get( + url, + json={ + "secrets": { + "A": {"raw": "va", "computed": ""}, + "B": {"raw": "", "computed": "vb"}, + "EMPTY": {"raw": " ", "computed": " "}, + } + }, + ) + result = store.get_many(["A", "B", "EMPTY", "MISSING"]) + assert result == {"A": "va", "B": "vb"} + + # A second batch must reuse the same single mocked response (one request total). + assert store.get_many(["A"]) == {"A": "va"} + assert len(rsps.calls) == 2 # one per get_many call, not per key + def test_list_keys_with_prefix(self): import responses @@ -465,6 +540,58 @@ def test_list_keys_with_prefix(self): keys = store.list_keys(prefix="DB_") assert keys == ["DB_HOST", "DB_PORT"] + def test_get_url_encodes_special_characters_in_key(self): + """Keys with special chars (&, =, #, spaces, quotes) must be URL-encoded in the filter.""" + from urllib.parse import quote + + import responses + + from envault.stores import OnePasswordStore + + store = OnePasswordStore(token="fake", vault_id="v1") + base_url = "http://localhost:8080/v1/vaults/v1/items" + # Key with characters that break unencoded URLs + key = 'MY&KEY=WITH#SPECIAL "CHARS"' + encoded_key = quote(key, safe="") + filter_url = f"{base_url}?filter=title%20eq%20%22{encoded_key}%22" + + with responses.RequestsMock() as rsps: + items = [ + { + "title": key, + "fields": [{"purpose": "PASSWORD", "value": "secret_val"}], + } + ] + rsps.get(filter_url, json=items) + result = store.get(key) + assert result == "secret_val" + # Verify the request was made with the properly encoded URL + assert len(rsps.calls) == 1 + assert encoded_key in rsps.calls[0].request.url + + def test_delete_url_encodes_special_characters_in_key(self): + """delete() must also URL-encode keys with special characters.""" + from urllib.parse import quote + + import responses + + from envault.stores import OnePasswordStore + + store = OnePasswordStore(token="fake", vault_id="v1") + base_url = "http://localhost:8080/v1/vaults/v1/items" + key = "KEY/WITH/SLASHES&" + encoded_key = quote(key, safe="") + filter_url = f"{base_url}?filter=title%20eq%20%22{encoded_key}%22" + item_id = "item-del-special" + + with responses.RequestsMock() as rsps: + rsps.get(filter_url, json=[{"id": item_id, "title": key}]) + rsps.delete(f"{base_url}/{item_id}", status=204) + result = store.delete(key) + assert result is True + assert len(rsps.calls) == 2 + assert encoded_key in rsps.calls[0].request.url + # ── Store factory deeper tests ──────────────────────────────────────────────