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/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..eee5dcc 100644 --- a/src/envault/stores/__init__.py +++ b/src/envault/stores/__init__.py @@ -346,7 +346,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 +373,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_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_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..e03b9dc 100644 --- a/tests/test_stores_integration.py +++ b/tests/test_stores_integration.py @@ -465,6 +465,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 ──────────────────────────────────────────────