refactor: de-dup JWT verification, DB lifecycle, and keypair generation#59
Open
PenguinzTech wants to merge 1 commit into
Open
refactor: de-dup JWT verification, DB lifecycle, and keypair generation#59PenguinzTech wants to merge 1 commit into
PenguinzTech wants to merge 1 commit into
Conversation
Reusable-code pass (no behavior changes; every suite re-verified): - dns-server: extract the duplicated ES256/RS256 verification contract (alg allowlist, iss/aud, required exp/iat/tenant, fail-closed) from selective_router.py and resilience.py into app/utils/jwt_verify.py (verify_squawk_jwt). Authorization (zone visibility/team rules) stays with the callers. Existing 24 @patch-based regression tests unchanged. - manager client_config_service: replace 12 copies of the 'db = self._get_db() ... finally: db.close()' lifecycle with a _with_db decorator that injects the connection and guarantees close. - manager keypair generation: new app/utils/crypto.generate_ephemeral_ es256_keypair() used by config.py (Dev+Test blocks) and test helpers (conftest jwt_keypair, domain-jwt tests). client_config_service keeps a self-contained copy BY DESIGN — it is imported bare by the acceptance harness where app.* resolves to a different package (documented inline). - Fix 2 stale dns-server acceptance tests missed by the domain-JWT asymmetric migration (still decoding with the removed HS256 jwt_secret attr); expired-token test now uses a genuinely expired ES256 token. - Fix pre-existing tests/test_installer.py::test_check_admin_windows: patching ctypes.windll fails on Linux (attribute doesn't exist); attach a MagicMock with create=True. Verified: manager 142, dns-server 58, client-config acceptance 25, root unit 67 — all passing. flake8/pyflakes clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Reviewer's GuideCentralizes JWT verification in dns-server, factors out ephemeral ES256 keypair generation in manager, and deduplicates DB connection lifecycle in client_config_service, plus fixes a few test issues so behavior stays consistent but implementation is cleaner and less error‑prone. Sequence diagram for dns-server JWT verification via verify_squawk_jwtsequenceDiagram
actor User
participant DNSSelectiveRouter
participant JWTVerifier
User->>DNSSelectiveRouter: check_zone_permission(domain, token)
DNSSelectiveRouter->>JWTVerifier: verify_squawk_jwt(token, JWT_PUBLIC_KEY)
JWTVerifier-->>DNSSelectiveRouter: payload or None
alt payload is not None
DNSSelectiveRouter->>DNSSelectiveRouter: apply zone visibility/team rules
DNSSelectiveRouter-->>User: allow or deny
else payload is None
DNSSelectiveRouter-->>User: deny
end
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="manager/backend/app/services/client_config_service.py" line_range="87-95" />
<code_context>
+ return private_pem, public_pem
+
+
+def _with_db(method):
+ """Open a DB connection for the wrapped method and guarantee it closes.
+
+ The wrapped method receives the connection as its first argument after
+ ``self``. Centralizes the get/close lifecycle previously duplicated in
+ every method (``db = self._get_db()`` ... ``finally: db.close()``).
+ """
+ @wraps(method)
+ def wrapper(self, *args, **kwargs):
+ db = self._get_db()
+ try:
</code_context>
<issue_to_address>
**suggestion:** Consider typing `_with_db` to clarify the wrapped signature and return type.
Since `_with_db` is untyped while the decorated methods explicitly take a `DB` parameter, please add type hints for `method` and the wrapper’s return value (e.g., `Callable[..., Any]` or a more specific signature). This will make it clearer to type checkers and IDEs that the decorator preserves the original call signature apart from the injected `db`, and help avoid misuse.
Suggested implementation:
```python
from typing import Any, Callable
def _with_db(method: Callable[..., Any]) -> Callable[..., Any]:
"""Open a DB connection for the wrapped method and guarantee it closes.
The wrapped method receives the connection as its first argument after
``self``. Centralizes the get/close lifecycle previously duplicated in
every method (``db = self._get_db()`` ... ``finally: db.close()``).
"""
@wraps(method)
def wrapper(self, *args: Any, **kwargs: Any) -> Any:
db = self._get_db()
try:
return method(self, db, *args, **kwargs)
finally:
db.close()
return wrapper
```
1. If this file already has `from typing import ...` imports, merge `Any` and `Callable` into that existing import line instead of adding a new one.
2. If you prefer more precise typing and you have access to the `DB` type and are using Python 3.10+ / `typing_extensions`, you can refine this using `ParamSpec` and `Concatenate` to model the injected `db` parameter explicitly.
</issue_to_address>
### Comment 2
<location path="dns-server/app/utils/jwt_verify.py" line_range="36-45" />
<code_context>
+logger = logging.getLogger(__name__)
+
+
+def verify_squawk_jwt(token: str, public_key: Optional[str]) -> Optional[dict]:
+ """Verify a Squawk user JWT; return its payload or None (fail closed).
+
+ Args:
+ token: The presented bearer token.
+ public_key: PEM public key to verify with (caller's configured key).
+
+ Returns:
+ The verified payload dict, or None on any failure — unconfigured key,
+ bad signature/alg, expired, wrong iss/aud, missing/empty tenant.
+ """
+ if not public_key:
+ logger.error("JWT_PUBLIC_KEY not configured; denying access")
+ return None
+
+ if not token:
+ return None
+
</code_context>
<issue_to_address>
**nitpick:** The `token` parameter type could be relaxed to `Optional[str]` to match the usage.
The function already treats a missing/empty token as valid input and returns early, so the signature should reflect that by using `token: Optional[str]`. This will better match the actual contract and prevent unnecessary type-checker issues when callers pass an optional token.
</issue_to_address>
### Comment 3
<location path="tests/test_installer.py" line_range="28-37" />
<code_context>
def test_check_admin_windows(self):
"""Test admin check on Windows"""
+ import ctypes
+ from unittest.mock import MagicMock
+
+ # ctypes.windll only exists on Windows — attach a mock (create=True)
+ # so the Windows code path is exercisable on Linux/macOS CI.
+ windll = MagicMock()
+ windll.shell32.IsUserAnAdmin.return_value = 1
with patch('platform.system', return_value='Windows'):
- with patch('ctypes.windll.shell32.IsUserAnAdmin', return_value=1):
+ with patch.object(ctypes, 'windll', windll, create=True):
installer = SquawkInstaller()
assert installer.is_admin is True
</code_context>
<issue_to_address>
**suggestion (testing):** Add a complementary test for the non-admin Windows path
To fully cover the Windows admin logic, add a second test where `IsUserAnAdmin` returns `0` and assert `installer.is_admin is False`, so both branches of the conditional are exercised and protected against regressions.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+87
to
+95
| def _with_db(method): | ||
| """Open a DB connection for the wrapped method and guarantee it closes. | ||
|
|
||
| The wrapped method receives the connection as its first argument after | ||
| ``self``. Centralizes the get/close lifecycle previously duplicated in | ||
| every method (``db = self._get_db()`` ... ``finally: db.close()``). | ||
| """ | ||
| @wraps(method) | ||
| def wrapper(self, *args, **kwargs): |
There was a problem hiding this comment.
suggestion: Consider typing _with_db to clarify the wrapped signature and return type.
Since _with_db is untyped while the decorated methods explicitly take a DB parameter, please add type hints for method and the wrapper’s return value (e.g., Callable[..., Any] or a more specific signature). This will make it clearer to type checkers and IDEs that the decorator preserves the original call signature apart from the injected db, and help avoid misuse.
Suggested implementation:
from typing import Any, Callable
def _with_db(method: Callable[..., Any]) -> Callable[..., Any]:
"""Open a DB connection for the wrapped method and guarantee it closes.
The wrapped method receives the connection as its first argument after
``self``. Centralizes the get/close lifecycle previously duplicated in
every method (``db = self._get_db()`` ... ``finally: db.close()``).
"""
@wraps(method)
def wrapper(self, *args: Any, **kwargs: Any) -> Any:
db = self._get_db()
try:
return method(self, db, *args, **kwargs)
finally:
db.close()
return wrapper- If this file already has
from typing import ...imports, mergeAnyandCallableinto that existing import line instead of adding a new one. - If you prefer more precise typing and you have access to the
DBtype and are using Python 3.10+ /typing_extensions, you can refine this usingParamSpecandConcatenateto model the injecteddbparameter explicitly.
Comment on lines
+36
to
+45
| def verify_squawk_jwt(token: str, public_key: Optional[str]) -> Optional[dict]: | ||
| """Verify a Squawk user JWT; return its payload or None (fail closed). | ||
|
|
||
| Args: | ||
| token: The presented bearer token. | ||
| public_key: PEM public key to verify with (caller's configured key). | ||
|
|
||
| Returns: | ||
| The verified payload dict, or None on any failure — unconfigured key, | ||
| bad signature/alg, expired, wrong iss/aud, missing/empty tenant. |
There was a problem hiding this comment.
nitpick: The token parameter type could be relaxed to Optional[str] to match the usage.
The function already treats a missing/empty token as valid input and returns early, so the signature should reflect that by using token: Optional[str]. This will better match the actual contract and prevent unnecessary type-checker issues when callers pass an optional token.
Comment on lines
28
to
37
| def test_check_admin_windows(self): | ||
| """Test admin check on Windows""" | ||
| import ctypes | ||
| from unittest.mock import MagicMock | ||
|
|
||
| # ctypes.windll only exists on Windows — attach a mock (create=True) | ||
| # so the Windows code path is exercisable on Linux/macOS CI. | ||
| windll = MagicMock() | ||
| windll.shell32.IsUserAnAdmin.return_value = 1 | ||
| with patch('platform.system', return_value='Windows'): |
There was a problem hiding this comment.
suggestion (testing): Add a complementary test for the non-admin Windows path
To fully cover the Windows admin logic, add a second test where IsUserAnAdmin returns 0 and assert installer.is_admin is False, so both branches of the conditional are exercised and protected against regressions.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #58. Seventh branch in the chain. Reusable-code pass — no behavior changes; every affected suite re-verified.
De-duplication
exp/iat/tenant, fail-closed) duplicated inselective_router.pyandresilience.pynow lives inapp/utils/jwt_verify.py(verify_squawk_jwt). Authorization (zone visibility / team rules) stays with the callers. The 24@patch-based auth-bypass regression tests run unchanged.client_config_service.pyhad 12 copies ofdb = self._get_db() … finally: db.close(); replaced with a_with_dbdecorator that injects the connection and guarantees close (bodies untouched → low-risk diff).app/utils/crypto.generate_ephemeral_es256_keypair()consumed byconfig.py(Dev+Test blocks) and test helpers (conftestjwt_keypair, domain-JWT tests).client_config_servicekeeps a self-contained copy by design — it's imported bare by the acceptance harness whereapp.*resolves to a different package (rationale documented inline; discovered the hard way when the shared import broke all 25 acceptance tests).Also fixes
jwt_secretattribute); the expired-token test now uses a genuinely expired ES256 token instead of exercising alg rejection.tests/test_installer.py::test_check_admin_windowsfailure: patchingctypes.windllcan't resolve on Linux — mock attached withcreate=True.Verification
manager 142, dns-server 58, client-config acceptance 25, root unit 67 — all passing. flake8/pyflakes clean.
🤖 Generated with Claude Code
Summary by Sourcery
Deduplicate JWT verification, DB connection lifecycle, and ES256 keypair generation across manager and dns-server while keeping behavior unchanged and fixing related test issues.
New Features:
Bug Fixes:
Enhancements:
Tests: