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
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()
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")
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
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