Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/envault/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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",
}
Expand Down
22 changes: 19 additions & 3 deletions src/envault/backup.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
34 changes: 3 additions & 31 deletions src/envault/serve.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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()
Comment on lines +302 to +304

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Filter advertised methods by the active auth mode

Routing requests to this discovery endpoint exposes a misleading response whenever configured credentials do not match auth_mode. For example, with auth_mode="bearer" and only api_key configured (the setup used by the new test), /auth/info advertises api-key, but _check_auth() takes the bearer branch and rejects every X-API-Key request; the inverse occurs for auth_mode="api-key" with an API token. Report only methods the selected mode actually accepts so clients do not choose credentials that are guaranteed to fail.

Useful? React with 👍 / 👎.

elif path == "/secrets":
if not self._check_auth():
return
Expand Down
10 changes: 8 additions & 2 deletions src/envault/stores/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Comment on lines +351 to +352

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Escape quotes before URL-encoding 1Password filters

For keys containing a quotation mark—an input the new test explicitly claims to support—quote() only protects the URL transport. The server decodes %22 before parsing the filter, turning a key such as MY "CHARS" into the malformed expression title eq "MY "CHARS""; consequently both get() and delete() fail against a real Connect server even though the URL-matching mock passes. Escape the key according to the filter string grammar before percent-encoding the complete filter expression.

Useful? React with 👍 / 👎.

if not items:
return None
item_list = items if isinstance(items, list) else items.get("items", [])
Expand All @@ -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", [])
Expand Down
37 changes: 37 additions & 0 deletions tests/test_auth.py
Original file line number Diff line number Diff line change
@@ -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
80 changes: 80 additions & 0 deletions tests/test_backup_manifest.py
Original file line number Diff line number Diff line change
@@ -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)]
12 changes: 3 additions & 9 deletions tests/test_cli_edge_cases.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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']}"
)
Expand All @@ -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}"
25 changes: 25 additions & 0 deletions tests/test_serve.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 ──────────────────────────────────────────────────

Expand Down
52 changes: 52 additions & 0 deletions tests/test_stores_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -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&AMP"
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 ──────────────────────────────────────────────

Expand Down
Loading