-
Notifications
You must be signed in to change notification settings - Fork 1
fix: OAuth2 token encoding and /auth/info route #49
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
4a882b5
2185149
f1547b1
526553d
343b88f
0831ab7
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
For keys containing a quotation mark—an input the new test explicitly claims to support— Useful? React with 👍 / 👎. |
||
| 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", []) | ||
|
|
||
| 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 |
| 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)] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Routing requests to this discovery endpoint exposes a misleading response whenever configured credentials do not match
auth_mode. For example, withauth_mode="bearer"and onlyapi_keyconfigured (the setup used by the new test),/auth/infoadvertisesapi-key, but_check_auth()takes the bearer branch and rejects everyX-API-Keyrequest; the inverse occurs forauth_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 👍 / 👎.