Harden subscription and admin security - #756
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughThe pull request adds bounded security settings, subscription and group access checks, synchronized cleanup, NATS-based multi-worker coordination, WireGuard reconciliation changes, dashboard variable support, and expanded tests and workflow updates. ChangesSecurity hardening and access control
Distributed runtime coordination
Dashboard and supporting updates
Estimated code review effort: 5 (Critical) | ~100 minutes Sequence Diagram(s)sequenceDiagram
participant Worker
participant AppFactory
participant NATS
participant Leader
participant NodeManager
Worker->>AppFactory: Start application
AppFactory->>NATS: Validate multi-worker availability
AppFactory->>Leader: Start role-scoped election
Leader->>NATS: Acquire or renew lease
Leader-->>AppFactory: Leadership state
AppFactory->>NodeManager: Start coordinated node services
NodeManager->>NATS: Publish node synchronization action
NATS-->>NodeManager: Deliver action to other workers
Estimated code review effort: 5 (Critical) | ~100 minutes Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull request overview
This PR hardens subscription access control and administrative security across API, background jobs, webhook notifications, and deployment artifacts, with accompanying tests and a targeted migration to reduce default network exposure.
Changes:
- Enforce subscription eligibility checks across config/raw/info/apps/header paths and bound subscription usage query ranges.
- Require expiring admin JWTs and redact sensitive subscription credentials from webhook payloads.
- Revoke deleted/expired user credentials on nodes during scheduled/manual cleanup, bind seeded client listeners to loopback, and safely quote installer paths in the systemd unit.
Reviewed changes
Copilot reviewed 12 out of 12 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_security_hardening.py | Adds focused regression tests for the new security hardening behaviors. |
| tests/api/test_user.py | Adds an API-level test ensuring disabled users cannot download subscription configs while still rendering the subscription page safely. |
| install_service.sh | Quotes/escapes install paths for systemd unit generation and rejects control-character paths. |
| config.py | Adds validation bounds for JWT expiry minutes and user cleanup autodelete days. |
| app/utils/jwt.py | Requires exp/iat/sub on admin JWTs and always sets an expiry when creating admin tokens. |
| app/operation/user.py | Updates manual expired-user deletion flow to revoke users on nodes and handle dry-run earlier. |
| app/operation/subscription.py | Adds eligibility gating for subscription output paths and enforces a maximum usage range window. |
| app/notification/webhook/init.py | Redacts subscription_url and proxy_settings from webhook-encoded user payloads. |
| app/models/user.py | Bounds auto_delete_in_days at the model level to prevent overflow/abuse. |
| app/jobs/remove_expired_users.py | Revokes deleted users on nodes during scheduled expiry cleanup. |
| app/db/migrations/versions/a8c2d491e705_bind_xray_client_inbounds_to_loopback.py | Migration updates seeded Xray subscription templates to listen on loopback. |
| app/db/crud/user.py | Adds an upper bound to persisted auto-delete values to keep datetime arithmetic safe. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
app/notification/webhook/__init__.py (1)
131-134: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winAdd bulk webhook redaction coverage.
The supplied security test covers
notify, butbulk_notifyhas a separate enqueue path. Add a test with multiple users and assert that each queued payload omitssubscription_urlandproxy_settings.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/notification/webhook/__init__.py` around lines 131 - 134, Add test coverage for bulk_notify using multiple messages/users, exercising its separate enqueue path and asserting every queued payload excludes subscription_url and proxy_settings. Reuse the existing webhook notification test fixtures and mocking patterns, and verify all messages are enqueued with the sensitive fields redacted.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@app/db/migrations/versions/a8c2d491e705_bind_xray_client_inbounds_to_loopback.py`:
- Around line 31-37: The migration’s exact string replacement misses valid JSON
whitespace variants of the listen field. Update the migration logic around the
client_templates content replacement to parse each template as JSON and set
every inbounds[].listen value from 0.0.0.0 to 127.0.0.1, then serialize the
modified document while preserving unaffected content and handling the validated
template format.
In `@app/operation/subscription.py`:
- Around line 91-96: Update user_subscription_by_user to call await
self.require_config_eligible(user) immediately after validated_user, ensuring
disabled, expired, and limited users are rejected before configuration
generation. Since user_subscription_by_id delegates to this method, add a
regression test covering rejection through the by-id path.
In `@app/operation/user.py`:
- Around line 1594-1596: Update both cleanup paths in app/operation/user.py
lines 1594-1596 and app/jobs/remove_expired_users.py line 19 to retain the tasks
created by sync_remove_user() and await them before deleting or reporting
deletion completion. Propagate or handle any dispatch failures so cleanup does
not return success when node removal fails; apply the same change at both sites.
In `@install_service.sh`:
- Around line 8-12: Update the INSTALL_DIR assignment to capture the complete
canonical output of pwd -P without command-substitution newline stripping,
preserving any trailing newline for the control-character validation. Also
detect and abort when canonicalization fails, before using INSTALL_DIR to
generate escaped systemd paths.
---
Nitpick comments:
In `@app/notification/webhook/__init__.py`:
- Around line 131-134: Add test coverage for bulk_notify using multiple
messages/users, exercising its separate enqueue path and asserting every queued
payload excludes subscription_url and proxy_settings. Reuse the existing webhook
notification test fixtures and mocking patterns, and verify all messages are
enqueued with the sensitive fields redacted.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 6d2148f2-ea37-4730-8e31-46eda30fed80
📒 Files selected for processing (12)
app/db/crud/user.pyapp/db/migrations/versions/a8c2d491e705_bind_xray_client_inbounds_to_loopback.pyapp/jobs/remove_expired_users.pyapp/models/user.pyapp/notification/webhook/__init__.pyapp/operation/subscription.pyapp/operation/user.pyapp/utils/jwt.pyconfig.pyinstall_service.shtests/api/test_user.pytests/test_security_hardening.py
b9bafa2 to
209a3db
Compare
There was a problem hiding this comment.
Actionable comments posted: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
.github/workflows/test-database-migrations.yml (1)
16-16: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winAdd an explicit
permissionsblock.The workflow declares no
permissions:key, so every job inherits the repository default token scopes. These jobs only run migrations and tests. They need no write access to repository contents.Add a least-privilege declaration at the workflow level.
🛡️ Proposed fix
+permissions: + contents: read + jobs: test-sqlite:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In @.github/workflows/test-database-migrations.yml at line 16, Add a workflow-level permissions block immediately under the top-level jobs configuration in the migration workflow, granting only the minimal read access required by migration and test jobs and leaving repository contents without write access.Source: Linters/SAST tools
app/node/__init__.py (1)
60-72: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPrevent sibling workers from remote-stopping the old node on
update_node.
remove_nodealready usesremote_stop=Falseforremoveanddisconnectsync messages, butapp/node/manager_sync.py:51still callsnode_manager.update_node(db_node)forupsert.update_node()stops the old node withremote_stop=True, so each sibling worker can send a duplicate remotestop()for the same core. Addremote_stopsupport toupdate_node()and passremote_stop=Falsefrom theupserthandler; the originator should be responsible for remote stops.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/__init__.py` around lines 60 - 72, Add a remote_stop parameter to update_node and pass it through to _shutdown_node, preserving the current default behavior for existing callers. In the upsert handler of manager_sync, call update_node with remote_stop=False so sibling workers do not issue duplicate remote stops; the originating worker remains responsible for remote shutdowns.
🧹 Nitpick comments (15)
app/nats/router.py (1)
15-17: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReuse the shared helpers instead of duplicating the predicate.
_router_enabledrepeats the exact expression ofapp/nats/__init__.pyis_multi_worker()combined withis_nats_enabled().app/core/hosts.pyandapp/core/manager.pyalready compose the two helpers. Keep one definition so the gates cannot diverge.♻️ Proposed refactor
-from config import nats_settings, runtime_settings, server_settings +from app.nats import is_multi_worker, is_nats_enabled +from config import nats_settings logger = get_logger("nats-router") def _router_enabled() -> bool: - multi_worker = runtime_settings.role.requires_nats or server_settings.workers > 1 - return nats_settings.enabled and multi_worker + return is_nats_enabled() and is_multi_worker()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/nats/router.py` around lines 15 - 17, Update _router_enabled to reuse the shared is_multi_worker() and is_nats_enabled() helpers from app.nats instead of recomputing runtime_settings and server_settings predicates; preserve the existing combined gate behavior and remove the duplicated local expression.app/nats/leader.py (1)
29-30: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReplace the module-level
assertwith an explicit check.Python removes
assertstatements when the interpreter runs with-O. The timing invariant then disappears silently. Raise aValueErrorinstead, or move the constraint into a unit test.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/nats/leader.py` around lines 29 - 30, Replace the module-level assert validating HEARTBEAT_INTERVAL, HEARTBEAT_MAX_RETRIES, HEARTBEAT_RETRY_DELAY, and DEFAULT_LEASE_SECONDS with an explicit runtime check that raises ValueError when the lease timing invariant is violated. Keep the existing inequality and validation at module initialization so it remains enforced under optimized Python execution.app/node/manager_sync.py (1)
46-72: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winThe ORM instance is used after the session closes.
Both the
upsertbranch and theconnectbranch loaddb_nodeinsideasync with GetDB() as dband then use it after the block exits.GetDB.__aexit__closes the session, which detaches the instance.The code works today because
load_usage_logs=Falseavoids relationship access, no commit expires the attributes, and every attribute read afterwards is an already-loaded column. It breaks if anyone later adds a relationship read, for exampledb_node.core_config, insidenode_manager.update_nodeorconnect_node. The failure isDetachedInstanceErrorat runtime on sibling workers only.Extend the session scope over the calls that consume
db_node.♻️ Proposed refactor for the `connect` branch
async with GetDB() as db: db_node = await get_node_by_id(db, node_id, load_usage_logs=False) if db_node is None or db_node.status in (NodeStatus.disabled, NodeStatus.limited): return core_id = db_node.core_config_id or 1 cores_by_id, users_by_core = await NodeOperation._get_core_users_map(db, {core_id}) core = cores_by_id.get(core_id) users = users_by_core.get(core_id, []) - try: - await node_manager.update_node(db_node) - except Exception: - logger.exception("Node sync connect update_node failed for node_id=%s", node_id) - return - await NodeOperation.connect_node(db_node, core, users) + try: + await node_manager.update_node(db_node) + except Exception: + logger.exception("Node sync connect update_node failed for node_id=%s", node_id) + return + await NodeOperation.connect_node(db_node, core, users) returnNote that this holds the database connection for the duration of the connect. If that is not acceptable, copy the required fields into a plain object instead.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/manager_sync.py` around lines 46 - 72, Extend the GetDB session scope in both the upsert and connect branches so db_node remains attached while node_manager.update_node and NodeOperation.connect_node consume it. Keep the existing status checks and error handling, and ensure the connect branch performs its node operations before the async context exits.app/jobs/node_checker.py (2)
252-262: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAdd startup jitter to the per-worker interval loop.
Every uvicorn worker runs
initialize_nodesat approximately the same moment, and Lines 287-292 start one_interval_loopper worker. All workers therefore runnode_health_checkfor all nodes in lockstep on every interval.NODE_CHECK_SEMcaps concurrency to 5 inside one worker, but it does not coordinate across workers, so the effective burst is 5 × worker count against the nodes and the database.A random initial delay spreads the load.
♻️ Proposed refactor
async def _interval_loop(coro, seconds: float, name: str): """Run node maintenance on every worker (APScheduler may be leader-only).""" + # Stagger workers so all uvicorn processes do not check in lockstep. + await asyncio.sleep(random.uniform(0, min(seconds, 30))) while True: try: await coro() except Exception as exc: logger.error("Node loop %s failed: %s", name, exc) await asyncio.sleep(seconds)Add
import randomat the top of the file.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/jobs/node_checker.py` around lines 252 - 262, Add startup jitter to _interval_loop by importing random and awaiting a random initial delay before entering its existing while loop, using a bounded delay appropriate to the interval (seconds). Preserve the current recurring execution, exception logging, and sleep behavior after the initial stagger.
137-162: 🚀 Performance & Scalability | 🔵 TrivialConsider the KV read cost of the unconditional lifecycle fetch.
Line 139 calls
node.get_lifecycle_state()for every node on every health check, before the branch that needs it. In multi-worker mode this resolves to a NATS KV read. Lines 285-292 start a health loop on every uvicorn worker, so the read count is nodes × workers per interval.The value is also used at Lines 174-175 and 199-200, so it cannot simply move inside the
NOT_CONNECTEDbranch. Consider adding a metric for lifecycle KV read volume, and confirm thatjob_settings.core_health_check_intervalis sized for the node count you expect.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/jobs/node_checker.py` around lines 137 - 162, Instrument the lifecycle-state retrieval around node.get_lifecycle_state() with a metric tracking KV read volume, including node/worker context if available. Keep the shared state available for its later uses at the referenced lifecycle checks, and review or validate core_health_check_interval against expected node and uvicorn worker counts so the resulting read rate is acceptable.app/operation/node.py (4)
321-332: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAdd a short delay before the 409 attach retry.
The 409 response means another worker holds the lifecycle lease and the start is still in progress. The retry at Line 323 runs immediately, so the remote core is very likely still starting and
pg_node.info()fails again. The node is then recorded aserrorat Line 338 even though the start succeeds moments later.A short sleep before the retry raises the success rate.
♻️ Proposed refactor
if e.code == 409: # Another worker holds the lifecycle lease; try attach once more. + await asyncio.sleep(1) attached = await NodeOperation._attach_if_running(pg_node, db_node.name)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/operation/node.py` around lines 321 - 332, In the 409 branch of the node start flow, add a short asynchronous delay before invoking NodeOperation._attach_if_running. Keep the existing single retry and success response unchanged after the delay, while preserving the current error handling when attachment still fails.
625-631: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueClear bridge memory after the siblings have removed the node.
_remove_node_localclears the shared lifecycle and user-sync keys, and_remove_node_syncpublishes theremovemessage afterwards. Sibling workers still hold the node object at that moment. A sibling health check running in that window callsupdate_observed_lifecycle, andNatsNodeLifecycleCoordinator.update_observedrecreates the lifecycle document when the document is missing andexpected_epochisNone. The sibling's ownclear_bridge_memory_for_nodethen removes it again, so the state converges, but a stale lifecycle document can exist briefly for a deleted node.Publishing
removebefore clearing shared memory narrows the window.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/operation/node.py` around lines 625 - 631, Update _remove_node_sync to publish the "remove" message immediately after _remove_node_local’s node removal step, then clear bridge memory only after sibling workers have processed the removal; adjust _remove_node_local or inline its cleanup so clear_bridge_memory_for_node runs after publish_node_sync while preserving the existing node_manager removal.
704-709: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftBatch the bulk connect fan-out into one message.
_connect_nodes_bulk_syncpublishes oneconnectmessage per node. Each sibling worker handles every message independently inhandle_node_message, andapp/node/manager_sync.pyLine 58 opens a freshGetDB()session and calls_get_core_users_mapfor each one. For N nodes and W sibling workers, a bulk restart produces N messages and N database sessions per worker.
_connect_nodes_bulk_remoteat Line 713 already uses a batched payload withnode_ids. Use the same shape here so siblings resolve cores and users once.♻️ Proposed refactor sketch
async def _connect_nodes_bulk_sync(self, db: AsyncSession, nodes: list[Node]) -> None: await self._connect_nodes_bulk_local(db, nodes) - for node in nodes: - if node is not None and node.status not in (NodeStatus.disabled, NodeStatus.limited): - await publish_node_sync("connect", node.id) + node_ids = [ + node.id + for node in nodes + if node is not None and node.status not in (NodeStatus.disabled, NodeStatus.limited) + ] + if node_ids: + await publish_node_sync_bulk("connect_bulk", node_ids)This requires a
connect_bulkbranch inhandle_node_messagethat resolves cores and users once for the whole set.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/operation/node.py` around lines 704 - 709, Update _connect_nodes_bulk_sync to publish one connect_bulk message containing the eligible node IDs instead of one connect message per node, matching _connect_nodes_bulk_remote’s payload shape. Add the corresponding connect_bulk branch in handle_node_message and resolve cores and users once for the full node set before processing it.
256-258: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winRaise the log level for attach failures.
_attach_if_runningcatches every exception and logs atdebug. The caller cannot distinguish "the node is not running" from "the attach RPC failed". The 409 recovery path at Line 321 depends on this method succeeding, so a persistent attach failure produces a silent reconnect loop with no visible cause at the default log level.Log at
warningwhen the failure occurs afterpg_node.info()returned a valid response, and keepdebugfor the expected not-running case.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/operation/node.py` around lines 256 - 258, Update _attach_if_running to distinguish expected not-running failures from attach RPC failures: retain debug logging when pg_node.info() does not return a valid response, but log at warning when an exception occurs after a valid pg_node.info() response and the attach operation fails. Preserve the existing exception handling and return None behavior.app/node/nats_memory.py (2)
255-297: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winLog CAS exhaustion in
try_acquire.
try_acquirereturnsNonefor two different conditions: another worker holds a live lease, and 32 CAS attempts all conflicted.release,heartbeat, andupdate_observedeach log a warning when their CAS loop is exhausted.try_acquiredoes not. A conflict storm is then invisible in the logs and appears as normal lease contention.♻️ Proposed fix
if await kv_cas_json(self._kv, key, doc, rev): return lease + logger.warning("Lifecycle try_acquire CAS exhausted for node_id=%s key=%s", node_id, key) return None🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/nats_memory.py` around lines 255 - 297, Update try_acquire’s CAS retry path to emit a warning when all 32 attempts fail, distinguishing CAS exhaustion from the existing early return for an active lease. Use the same warning-log convention as release, heartbeat, and update_observed, then return None as before.
119-128: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winSequential per-key KV round trips across
NatsUserSyncStore. Every method that touches multiple keys awaits one KV operation at a time. Eachkv_put_jsoncosts at least two round trips, and each guarded delete costs two. For a node with N pending users, a full sync or a full clear costs O(N) sequential NATS round trips on the user-sync path. The shared root cause is the absence of a bounded-concurrency helper for per-key operations; add one and reuse it in all four methods.
app/node/nats_memory.py#L119-L128: gather thekv_put_jsoncalls inenqueue_usersunder a semaphore, after validating all value sizes up front.app/node/nats_memory.py#L195-L206: gather the deletes inack_usersunder the same semaphore, and drop the preceding read as noted separately.app/node/nats_memory.py#L208-L223: gather the per-item requeue-then-delete pairs inrequeue_users, keeping the write-before-delete order inside each pair.app/node/nats_memory.py#L225-L241: gather the pending and claimed deletes inclearunder the same semaphore.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/nats_memory.py` around lines 119 - 128, Sequential KV operations in NatsUserSyncStore need bounded concurrency. Add a shared semaphore-based per-key operation helper and reuse it in app/node/nats_memory.py:119-128 (enqueue_users: validate all value sizes first, then gather kv_put_json calls), app/node/nats_memory.py:195-206 (ack_users: gather deletes and remove the preceding read), app/node/nats_memory.py:208-223 (requeue_users: gather per-item operations while preserving write-before-delete order), and app/node/nats_memory.py:225-241 (clear: gather pending and claimed deletes).app/node/__init__.py (1)
49-58: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winLog the failure in
_shutdown_node.The
except Exception: passblock discards every error fromset_healthandstop(). A failed remotestop()leaves the remote core running and produces no record. This PR changes remote-stop behavior across workers, so a silent swallow here hides exactly the class of lifecycle bug the change introduces.♻️ Proposed refactor
try: await node.set_health(Health.INVALID) if remote_stop: await node.stop() - except Exception: - pass + except Exception as exc: + self.logger.debug("Node shutdown failed (remote_stop=%s): %s", remote_stop, exc)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/__init__.py` around lines 49 - 58, Update `_shutdown_node` to log exceptions raised by `set_health` or the conditional remote `stop()` call instead of silently swallowing them; preserve the existing shutdown flow and exception handling while including enough context to identify the affected node and operation.app/nats/kv_cas.py (3)
108-127: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAlign the test double with NATS KV semantics.
Two divergences exist between
MemoryCasKvand a real NATS KV bucket:
- Line 108:
deletereturnsFalsefor a missing key. NATS KVdeletewrites a delete marker and does not signal "missing". Tests that assert aFalsereturn will not match production behavior.- Line 124:
keys(filters=...)uses substring matching (f in key). NATS treats filters as subject filters with.token wildcards. A test that passes a filter passes for the wrong reason.Additionally,
getafterdeleteraisesKeyNotFoundErrorhere, while NATS raisesKeyDeletedError.kv_get_jsonhandles both, so the current callers are unaffected.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/nats/kv_cas.py` around lines 108 - 127, Update MemoryCasKv.delete to match NATS semantics by recording a delete marker for missing or existing keys instead of returning False, and ensure subsequent get raises KeyDeletedError for deleted keys. Replace substring matching in keys with NATS subject-filter matching using dot-delimited tokens and wildcards, while preserving NoKeysError when nothing matches.
52-58: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueAdd a small backoff between CAS retries.
kv_put_jsonretries 32 times with no delay. Concurrent writers on the same key produce a tight loop of KV round trips. A short randomized sleep reduces contention and NATS load.♻️ Proposed refactor
+import asyncio +import random + async def kv_put_json(kv: CasKv, key: str, value: dict[str, Any]) -> None: """Upsert JSON with CAS retries (latest value wins).""" - for _ in range(32): + for attempt in range(32): _, rev = await kv_get_json(kv, key) if await kv_cas_json(kv, key, value, rev): return + await asyncio.sleep(min(0.05, 0.002 * (attempt + 1)) * random.random()) raise RuntimeError(f"failed to put NATS KV key={key} after CAS retries")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/nats/kv_cas.py` around lines 52 - 58, Add a short randomized asynchronous sleep between failed CAS attempts in kv_put_json, while preserving the immediate return on success and the existing retry limit and error behavior. Use the module’s established async timing/randomness utilities if available.
61-66: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winUse a server-side key-listing filter.
kv_list_keyscalls the protocol method without a prefix, then filters every bucket key in Python. Replace it withnats-pykey listing usinglist_keys()or keys support that matchesf"{prefix}>", then updateMemoryCasKvto implement the same semantics instead of substring filtering.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/nats/kv_cas.py` around lines 61 - 66, Update kv_list_keys to request prefix-filtered keys server-side using the nats-py list_keys or equivalent support with the f"{prefix}>" pattern, while preserving the empty result for NoKeysError. Update MemoryCasKv to implement the same prefix semantics and avoid substring-based filtering.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/core/hosts.py`:
- Around line 74-75: Normalize inbound_config.get("finalmask") through the
FinalMask model before building fms, so model and dict inputs serialize with the
same deterministic key set rather than omitting unset fields inconsistently.
Update the final_mask_settings/fms serialization flow while preserving compact
JSON in finalmask_link, and ensure cached NATS KV payloads are refreshed or
invalidated when hosts are re-added.
In `@app/nats/leader.py`:
- Around line 60-71: Update _ensure_kv so that when get_or_create_kv_bucket
returns None, it closes the client stored in _nc before returning, while
preserving the existing successful bucket path.
In `@app/node/__init__.py`:
- Around line 76-82: Update the node shutdown lifecycle around
NodeManager.__init__ and remove_node: retain each asyncio.create_task result in
a self._pending_shutdowns set, and register completion cleanup so finished tasks
are removed. Ensure _remove_node_local does not clear bridge memory until the
corresponding shutdown task has completed, preventing lifecycle state from being
recreated by late stop writes.
In `@app/node/nats_memory.py`:
- Around line 345-353: Update the docstring of has_active_lease to describe any
unexpired lifecycle lease, without implying ownership by another worker.
Preserve the existing lease lookup and expiration behavior used by
process_node_health_check.
- Around line 436-444: Update shutdown_bridge_memory to serialize shutdown with
_init_lock and suppress exceptions from _nc.close(), matching
ensure_bridge_memory’s guarded close behavior. Hold the lock while closing the
client and clearing _nc, _user_sync_kv, _lifecycle_kv, _user_sync_store, and
_lifecycle_coordinator so a concurrent ensure_bridge_memory cannot repopulate
them during shutdown.
- Around line 155-191: In claim_users, move the time.time() call inside the
pending-key loop so each claimed_value gets a fresh expires_at timestamp.
Replace the comment near kv_cas_json with one identifying delete(pending_key,
last=rev) as the revision-based exclusion that allows only one worker to claim a
pending entry; preserve that delete guard and existing cleanup behavior.
- Around line 119-128: Update enqueue_users to build and validate every
deduplicated user payload with _ensure_value_size before performing any writes,
so validation errors leave the batch unwritten. Then replace the sequential
kv_put_json loop with bounded-concurrency writes, reusing the existing key/value
construction and ensuring all scheduled writes are awaited before returning.
In `@app/operation/node.py`:
- Around line 260-266: Update _start_or_attach_node so it delegates the
lifecycle-state decision to _attach_if_running instead of restricting attachment
to observed HEALTHY. Remove the HEALTHY-only guard and invoke _attach_if_running
for the relevant state, preserving its existing predicate so STARTING nodes are
attached before attempting pg_node.start.
- Around line 644-664: Update connect_single in the bulk connection flow to
catch unexpected exceptions from node_manager.update_node and connect_node,
converting each failure into a per-node error result instead of allowing
asyncio.gather to abort. Preserve the existing NodeAPIError details where
available, provide a safe error message for other exceptions, and ensure results
still reach bulk_update_node_status so every node is processed and notified.
In
`@dashboard/src/features/subscriptions/components/subscription-settings-schema.ts`:
- Line 105: Update the announce_url schema validation around announce_url so
non-empty values are validated as URLs after template expansion, while
preserving acceptance of empty strings and optional values. Reuse the existing
_format_announce_url() behavior or its shared validation mechanism to reject
malformed URLs and disallowed schemes before persistence and publication.
---
Outside diff comments:
In @.github/workflows/test-database-migrations.yml:
- Line 16: Add a workflow-level permissions block immediately under the
top-level jobs configuration in the migration workflow, granting only the
minimal read access required by migration and test jobs and leaving repository
contents without write access.
In `@app/node/__init__.py`:
- Around line 60-72: Add a remote_stop parameter to update_node and pass it
through to _shutdown_node, preserving the current default behavior for existing
callers. In the upsert handler of manager_sync, call update_node with
remote_stop=False so sibling workers do not issue duplicate remote stops; the
originating worker remains responsible for remote shutdowns.
---
Nitpick comments:
In `@app/jobs/node_checker.py`:
- Around line 252-262: Add startup jitter to _interval_loop by importing random
and awaiting a random initial delay before entering its existing while loop,
using a bounded delay appropriate to the interval (seconds). Preserve the
current recurring execution, exception logging, and sleep behavior after the
initial stagger.
- Around line 137-162: Instrument the lifecycle-state retrieval around
node.get_lifecycle_state() with a metric tracking KV read volume, including
node/worker context if available. Keep the shared state available for its later
uses at the referenced lifecycle checks, and review or validate
core_health_check_interval against expected node and uvicorn worker counts so
the resulting read rate is acceptable.
In `@app/nats/kv_cas.py`:
- Around line 108-127: Update MemoryCasKv.delete to match NATS semantics by
recording a delete marker for missing or existing keys instead of returning
False, and ensure subsequent get raises KeyDeletedError for deleted keys.
Replace substring matching in keys with NATS subject-filter matching using
dot-delimited tokens and wildcards, while preserving NoKeysError when nothing
matches.
- Around line 52-58: Add a short randomized asynchronous sleep between failed
CAS attempts in kv_put_json, while preserving the immediate return on success
and the existing retry limit and error behavior. Use the module’s established
async timing/randomness utilities if available.
- Around line 61-66: Update kv_list_keys to request prefix-filtered keys
server-side using the nats-py list_keys or equivalent support with the
f"{prefix}>" pattern, while preserving the empty result for NoKeysError. Update
MemoryCasKv to implement the same prefix semantics and avoid substring-based
filtering.
In `@app/nats/leader.py`:
- Around line 29-30: Replace the module-level assert validating
HEARTBEAT_INTERVAL, HEARTBEAT_MAX_RETRIES, HEARTBEAT_RETRY_DELAY, and
DEFAULT_LEASE_SECONDS with an explicit runtime check that raises ValueError when
the lease timing invariant is violated. Keep the existing inequality and
validation at module initialization so it remains enforced under optimized
Python execution.
In `@app/nats/router.py`:
- Around line 15-17: Update _router_enabled to reuse the shared
is_multi_worker() and is_nats_enabled() helpers from app.nats instead of
recomputing runtime_settings and server_settings predicates; preserve the
existing combined gate behavior and remove the duplicated local expression.
In `@app/node/__init__.py`:
- Around line 49-58: Update `_shutdown_node` to log exceptions raised by
`set_health` or the conditional remote `stop()` call instead of silently
swallowing them; preserve the existing shutdown flow and exception handling
while including enough context to identify the affected node and operation.
In `@app/node/manager_sync.py`:
- Around line 46-72: Extend the GetDB session scope in both the upsert and
connect branches so db_node remains attached while node_manager.update_node and
NodeOperation.connect_node consume it. Keep the existing status checks and error
handling, and ensure the connect branch performs its node operations before the
async context exits.
In `@app/node/nats_memory.py`:
- Around line 255-297: Update try_acquire’s CAS retry path to emit a warning
when all 32 attempts fail, distinguishing CAS exhaustion from the existing early
return for an active lease. Use the same warning-log convention as release,
heartbeat, and update_observed, then return None as before.
- Around line 119-128: Sequential KV operations in NatsUserSyncStore need
bounded concurrency. Add a shared semaphore-based per-key operation helper and
reuse it in app/node/nats_memory.py:119-128 (enqueue_users: validate all value
sizes first, then gather kv_put_json calls), app/node/nats_memory.py:195-206
(ack_users: gather deletes and remove the preceding read),
app/node/nats_memory.py:208-223 (requeue_users: gather per-item operations while
preserving write-before-delete order), and app/node/nats_memory.py:225-241
(clear: gather pending and claimed deletes).
In `@app/operation/node.py`:
- Around line 321-332: In the 409 branch of the node start flow, add a short
asynchronous delay before invoking NodeOperation._attach_if_running. Keep the
existing single retry and success response unchanged after the delay, while
preserving the current error handling when attachment still fails.
- Around line 625-631: Update _remove_node_sync to publish the "remove" message
immediately after _remove_node_local’s node removal step, then clear bridge
memory only after sibling workers have processed the removal; adjust
_remove_node_local or inline its cleanup so clear_bridge_memory_for_node runs
after publish_node_sync while preserving the existing node_manager removal.
- Around line 704-709: Update _connect_nodes_bulk_sync to publish one
connect_bulk message containing the eligible node IDs instead of one connect
message per node, matching _connect_nodes_bulk_remote’s payload shape. Add the
corresponding connect_bulk branch in handle_node_message and resolve cores and
users once for the full node set before processing it.
- Around line 256-258: Update _attach_if_running to distinguish expected
not-running failures from attach RPC failures: retain debug logging when
pg_node.info() does not return a valid response, but log at warning when an
exception occurs after a valid pg_node.info() response and the attach operation
fails. Preserve the existing exception handling and return None behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 0484732a-9e20-429c-b497-63e3b77b8a44
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (51)
.env.example.github/workflows/api-codeql.yml.github/workflows/build-dev.yml.github/workflows/build.yml.github/workflows/frontend-codeql.yml.github/workflows/test-database-migrations.ymlapp/app_factory.pyapp/core/hosts.pyapp/core/manager.pyapp/db/crud/wireguard.pyapp/jobs/node_checker.pyapp/jobs/record_usages.pyapp/nats/__init__.pyapp/nats/kv_cas.pyapp/nats/leader.pyapp/nats/message.pyapp/nats/router.pyapp/node/__init__.pyapp/node/manager_sync.pyapp/node/nats_memory.pyapp/operation/__init__.pyapp/operation/core.pyapp/operation/group.pyapp/operation/node.pyapp/operation/subscription.pyapp/operation/user.pyapp/operation/user_template.pyapp/routers/group.pyapp/subscription/share.pyconfig.pydashboard/public/statics/locales/en.jsondashboard/public/statics/locales/fa.jsondashboard/public/statics/locales/ru.jsondashboard/public/statics/locales/zh.jsondashboard/src/components/ui/variables-popover.tsxdashboard/src/features/hosts/dialogs/host-modal.tsxdashboard/src/features/subscriptions/components/subscription-general-settings-section.tsxdashboard/src/features/subscriptions/components/subscription-settings-schema.tsdashboard/src/features/users/components/action-buttons.tsxdashboard/src/pages/_dashboard.settings.general.tsxpyproject.tomlrole.pytests/api/test_core.pytests/api/test_user.pytests/test_connect_concurrency.pytests/test_create_app_nats_guard.pytests/test_group_access_unit.pytests/test_nats_leader_heartbeat.pytests/test_nats_leader_steal.pytests/test_nats_node_memory.pytests/test_node_manager_sync.py
🚧 Files skipped from review as they are similar to previous changes (2)
- app/operation/user.py
- app/operation/subscription.py
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
🛑 Comments failed to post (10)
app/core/hosts.py (1)
74-75: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash # Description: Trace finalmask_link producers and consumers, and the dict-source branch. set -euo pipefail rg -n -C 5 'finalmask_link|finalmask' --glob '*.py' rg -n -C 3 'finalmask' --glob '*.ts' --glob '*.tsx' --glob '*.json'Repository: PasarGuard/panel
Length of output: 224
🏁 Script executed:
#!/bin/bash set -euo pipefail echo "== files ==" git ls-files | sed -n '1,200p' echo "== hosts.py occurrences ==" rg -n -C 8 'finalmask_link|final_mask_settings|exclude_none|model_dump|ModelDump\(' -g '*.py' . echo "== python source outline candidates ==" for f in $(git ls-files '*.py' | xargs grep -l 'finalmask_link|final_mask_settings|HostManager|finalmask' 2>/dev/null || true); do echo "--- $f" wc -l "$f" ast-grep outline "$f" 2>/dev/null | sed -n '1,120p' || true done echo "== relevant file slices ==" python3 - <<'PY' import pathlib, subprocess names = subprocess.check_output(['git','ls-files','*.py'], text=True).splitlines() for n in names: if any(v in n for v in ['core/hosts.py','host','nats','config']): print('---', n) lines = pathlib.Path(n).read_text(errors='replace').splitlines() starts = [i for i,l in enumerate(lines,1) if 'finalmask' in l or 'HostManager' in l or 'nats' in l or 'finalmask_link' in l] for s in starts: lo=max(1,s-8); hi=min(len(lines),s+12) print(f'@@ lines {lo}-{hi}') for ii in range(lo,hi+1): print(f'{ii:4}: {lines[ii-1]}') PYRepository: PasarGuard/panel
Length of output: 50373
🏁 Script executed:
#!/bin/bash set -euo pipefail python3 - <<'PY' import json from dataclasses import dataclass, asdict from enum import Enum class FinalMaskTcpType(str, Enum): FRAGMENT = "fragment" REALM = "realm" `@dataclass` class FinalMaskTcpLayer: type: FinalMaskTcpType settings: dict class FinalMaskTcpSettings: def model_dump(self, **kwargs): return asdict(self) `@dataclass` class FinalMaskTcpLayerNew: type: FinalMaskTcpType settings: FinalMaskTcpSettings old_format = { "tcp": [{ "type": "fragment", "settings": { "packets": "tlshello", "length": "10-15", "interval": "5-5", } }] } new_format = FinalMaskTcpLayerNew( type=FinalMaskTcpType.FRAGMENT, settings=FinalMaskTcpSettings() ).model_dump() new_format["tcp"] = new_format.pop("settings").model_dump() old = FinalMaskTcpLayer( type=FinalMaskTcpType.FRAGMENT, settings=new_format["tcp"].copy() ) new = FinalMaskTcpLayerNew( type=FinalMaskTcpType.FRAGMENT, settings=new_format["tcp"].copy() ) print("old dict includes omitted fields:", repr(json.dumps(old_format, separators=(",", ":")))) print("old dict excludes omitted fields:", repr(json.dumps({k:v for k,v in old_format.items() if k != "tcp"} + [{"tcp": [{k:v for k,v in old_format["tcp"][0].items() if k != "length"}]}]), separators=(",", ":")))) PY echo "== app/core/hosts.py relevant sections ==" sed -n '1,140p' app/core/hosts.py sed -n '420,460p' app/core/hosts.py sed -n '600,640p' app/core/hosts.py echo "== SubscriptionInboundData definition ==" rg -n -C 20 'class SubscriptionInboundData|finalmask_link|finalmask' app/db/crud/core.py app/dRepository: PasarGuard/panel
Length of output: 643
🏁 Script executed:
#!/bin/bash set -euo pipefail python3 - <<'PY' # Behavioral probe of the changed producer branch. from enum import Enum import json from typing import Any class FinalMaskTcpType(str, Enum): FRAGMENT = "fragment" class FinalMaskTcpSettings: def model_dump(self, **kwargs): # Pydantic exclude_none=True removes None attributes; # this represents the model branch with None settings omitted. return {} class FinalMaskTcpLayer: type: FinalMaskTcpType settings: FinalMaskTcpSettings def model_dump(self, **kwargs): out: dict[str, Any] if kwargs.get("by_alias") is True: out = {} else: out = {} out["type"] = self.type.value if isinstance(self.type, Enum) else self.type sub = self.settings.model_dump(**kwargs) out["settings"] = sub if not kwargs.get("exclude_none") or any(v is not None for v in out.values()): pass def model_dump(self, *, by_alias: bool = False, exclude_none: bool = False, ...): kwargs = {k: v for k, v in globals().items() if k not in ("self", "kwargs", "...")} return self.model_dump(by_alias=by_alias, exclude_none=exclude_none) old_dict_branch = { "tcp": [{ "type": "fragment", "settings": { "packets": "tlshello", "length": "10-15", "interval": "5-5", } }] } class FinalMaskTcpSettingsWithNone: def model_dump(self, *, by_alias=False, exclude_none=False, **extra): sub = {"packets": "tlshello", "length": "10-15", "interval": "5-5"} if exclude_none is True: return {} return sub class FinalMaskTcpLayerWithNone: def model_dump(self, *, by_alias=False, exclude_none=False, **extra): sub = self.settings.model_dump(by_alias=by_alias, exclude_none=exclude_none, **extra) return {"settings": sub, "type": "fragment"} final_mask_settings_model_branch = FinalMaskTcpLayerWithNone(settings=FinalMaskTcpSettingsWithNone()) fms_model = final_mask_settings_model_branch.model_dump(by_alias=False, exclude_none=True) fms_dict = old_dict_branch print("model branch with exclude_none=True:", fms_model) print("dict branch with exclude_none=False:", fms_dict) print("model branch JSON:", json.dumps(fms_model, separators=(",", ":"))) print("dict branch JSON:", json.dumps(fms_dict, separators=(",", ":"))) PY echo "== app/core/hosts.py relevant sections ==" sed -n '62,122p' app/core/hosts.py sed -n '438,456p' app/core/hosts.py echo "== SubscriptionInboundData definition ==" rg -n -C 25 'class SubscriptionInboundData|finalmask_link|finalmask' app/db/crud/core.py app/db/crud/host.py app/models/subscription.pyRepository: PasarGuard/panel
Length of output: 376
🏁 Script executed:
#!/bin/bash set -euo pipefail echo "== deterministic key-set behavior ==" python3 - <<'PY' import json model_branch = {} # FinalMask model branch omits all None finalmask fields dict_branch = { "tcp": [{ "type": "fragment", "settings": { "packets": "tlshello", "length": "10-15", "interval": "5-5", } }] } for name, value in [("model branch", model_branch), ("dict branch", dict_branch)]: print(f"{name}: {json.dumps(value, separators=(',', ':'))}") PY echo "== app/core/hosts.py relevant sections ==" sed -n '62,122p' app/core/hosts.py sed -n '438,456p' app/core/hosts.py echo "== SubscriptionInboundData definition ==" rg -n -C 30 'class SubscriptionInboundData|finalmask_link|finalmask' app/db/crud/core.py app/models/subscription.pyRepository: PasarGuard/panel
Length of output: 10702
Ensure
finalmask_linkremains stable for both model and dict inputs.Line 72 checks
if final_mask_settings, and line 74 only callsmodel_dumpforFinalMask; if the inbound config suppliesfinalmaskas a dict,fmsstays a dict, but the model branch now omits missing finalmask fields from the JSON. Make the key set deterministic across both sources or normalizeinbound_config.get("finalmask")with pydantic before serialization. Also account for cached NATS KV state that can carry the pre-change payload until hosts are re-added.🧰 Tools
🪛 ast-grep (0.45.0)
[info] 74-74: use jsonify instead of json.dumps for JSON output
Context: json.dumps(fms, separators=(",", ":"))
Note: [CWE-116] Improper Encoding or Escaping of Output.(use-jsonify)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/core/hosts.py` around lines 74 - 75, Normalize inbound_config.get("finalmask") through the FinalMask model before building fms, so model and dict inputs serialize with the same deterministic key set rather than omitting unset fields inconsistently. Update the final_mask_settings/fms serialization flow while preserving compact JSON in finalmask_link, and ensure cached NATS KV payloads are refreshed or invalidated when hosts are re-added.app/nats/leader.py (1)
60-71: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Close the NATS client when bucket creation fails.
_ensure_kvassigns_ncbefore it creates the bucket. Ifget_or_create_kv_bucketreturnsNone,_kvstaysNonebut the connection stays open. The reclaim loop inapp/app_factory.pycallsstart_job_leadereveryHEARTBEAT_INTERVALseconds, sotry_become_leaderre-enters_ensure_kvand overwrites_ncwith a new client each time. Each overwrite leaks the previous connection.🔒️ Proposed fix to release the client on failure
_nc = await create_nats_client() if _nc is None: return None js = await get_jetstream_context(_nc) _kv = await get_or_create_kv_bucket(js, nats_settings.scheduler_leader_kv_bucket) + if _kv is None: + with contextlib.suppress(Exception): + await _nc.close() + _nc = None return _kv🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/nats/leader.py` around lines 60 - 71, Update _ensure_kv so that when get_or_create_kv_bucket returns None, it closes the client stored in _nc before returning, while preserving the existing successful bucket path.app/node/__init__.py (1)
76-82: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Retain the shutdown task and consider the ordering against bridge-memory clearing.
Two points:
asyncio.create_taskresult is not stored. The event loop keeps only a weak reference to a task, so a task can be garbage collected before it completes. Keep a reference until the task finishes._remove_node_localinapp/operation/node.pyLine 625 awaitsremove_nodeand then callsclear_bridge_memory_for_node. Because the shutdown runs detached, the lifecycle key is deleted whilestop()may still be running.NatsNodeLifecycleCoordinator.update_observedrecreates the document when the document is missing andexpected_epochisNone, so a late write can resurrect lifecycle state for a removed node.🛡️ Proposed fix for the task reference
async def remove_node(self, id: int, *, remote_stop: bool = True) -> None: async with self._lock.writer_lock: old_node: PasarGuardNode | None = self._nodes.pop(id, None) self._user_sync_locks.pop(id, None) # Do cleanup without holding the lock to avoid slow delete operations. - asyncio.create_task(self._shutdown_node(old_node, remote_stop=remote_stop)) + task = asyncio.create_task(self._shutdown_node(old_node, remote_stop=remote_stop)) + self._pending_shutdowns.add(task) + task.add_done_callback(self._pending_shutdowns.discard)Add
self._pending_shutdowns: set[asyncio.Task] = set()to__init__.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/__init__.py` around lines 76 - 82, Update the node shutdown lifecycle around NodeManager.__init__ and remove_node: retain each asyncio.create_task result in a self._pending_shutdowns set, and register completion cleanup so finished tasks are removed. Ensure _remove_node_local does not clear bridge memory until the corresponding shutdown task has completed, preventing lifecycle state from being recreated by late stop writes.app/node/nats_memory.py (4)
119-128: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
Enqueue users concurrently and validate sizes before writing.
Two concerns exist in
enqueue_users:
- The loop awaits
kv_put_jsononce per user. Each call performs at least onegetplus onecreate/updateround trip. A full sync of N users costs 2N sequential NATS round trips on the user-sync hot path._ensure_value_sizeraisesRuntimeErrorinside the loop. One oversized user aborts the call after part of the batch is already written. The remaining users are dropped without a retry path.Validate all values first, then write with a bounded concurrency.
♻️ Proposed refactor
async def enqueue_users(self, node_id: str, users: list[User]) -> None: if not users: return # Latest payload per email wins (dedupe across the batch first). by_email = {user.email: user for user in users} - for email, user in by_email.items(): - key = self._pending_key(node_id, email) - value = {"email": email, "user": _b64_user(user)} - self._ensure_value_size(key, value) - await kv_put_json(self._kv, key, value) + entries: list[tuple[str, dict[str, Any]]] = [] + for email, user in by_email.items(): + key = self._pending_key(node_id, email) + value = {"email": email, "user": _b64_user(user)} + self._ensure_value_size(key, value) + entries.append((key, value)) + + sem = asyncio.Semaphore(16) + + async def _write(key: str, value: dict[str, Any]) -> None: + async with sem: + await kv_put_json(self._kv, key, value) + + await asyncio.gather(*(_write(key, value) for key, value in entries))📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async def enqueue_users(self, node_id: str, users: list[User]) -> None: if not users: return # Latest payload per email wins (dedupe across the batch first). by_email = {user.email: user for user in users} entries: list[tuple[str, dict[str, Any]]] = [] for email, user in by_email.items(): key = self._pending_key(node_id, email) value = {"email": email, "user": _b64_user(user)} self._ensure_value_size(key, value) entries.append((key, value)) sem = asyncio.Semaphore(16) async def _write(key: str, value: dict[str, Any]) -> None: async with sem: await kv_put_json(self._kv, key, value) await asyncio.gather(*(_write(key, value) for key, value in entries))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/nats_memory.py` around lines 119 - 128, Update enqueue_users to build and validate every deduplicated user payload with _ensure_value_size before performing any writes, so validation errors leave the batch unwritten. Then replace the sequential kv_put_json loop with bounded-concurrency writes, reusing the existing key/value construction and ensuring all scheduled writes are awaited before returning.
155-191: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Correct the exclusion comment and refresh the lease timestamp per claim.
Two points in
claim_users:
- Line 177 states that create-only prevents two workers from claiming the same token key. That is not the mechanism.
tokenembeds a freshuuid4()per iteration, sokv_cas_json(self._kv, claimed_key, claimed_value, 0)can never collide across workers. The real exclusion isawait self._kv.delete(pending_key, last=rev)on Line 180, where only one worker wins the revision. Keep that delete, and fix the comment so a later change does not remove the guard that actually works.- Line 155 captures
nowonce before the loop. Every claim in the batch derivesexpires_atfrom that single timestamp. A slow loop shortens the effective lease for the later claims. Read the clock per iteration.♻️ Proposed fix
result: list[ClaimedUser] = [] - now = time.time() for pending_key in await kv_list_keys(self._kv, self._pending_prefix(node_id)): @@ token = f"{worker_id}:{uuid4()}" claimed_key = self._claimed_key(node_id, token) claimed_value = { "token": token, "email": email, "user": user_b64, - "expires_at": now + lease_seconds, + "expires_at": time.time() + lease_seconds, } self._ensure_value_size(claimed_key, claimed_value) try: - # Create-only so two workers cannot claim into the same token key. + # Token is unique per iteration, so this create always succeeds. + # Exclusion comes from the revision-guarded delete of pending_key below: + # only one worker can delete at revision `rev`. if not await kv_cas_json(self._kv, claimed_key, claimed_value, 0): continue await self._kv.delete(pending_key, last=rev)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/nats_memory.py` around lines 155 - 191, In claim_users, move the time.time() call inside the pending-key loop so each claimed_value gets a fresh expires_at timestamp. Replace the comment near kv_cas_json with one identifying delete(pending_key, last=rev) as the revision-based exclusion that allows only one worker to claim a pending entry; preserve that delete guard and existing cleanup behavior.
345-353: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the
has_active_leasedocstring.The docstring says "another worker still holds an unexpired lifecycle lease". The method does not compare the lease owner against the calling worker. It returns
Truefor a lease held by the calling worker as well.process_node_health_checkinapp/jobs/node_checker.pyrelies on this to skip a reconnect while a start is in flight, including a start owned by the same process, so the behavior is correct. Only the wording is wrong.📝 Proposed fix
async def has_active_lease(self, node_id: str) -> bool: - """True when another worker still holds an unexpired lifecycle lease.""" + """True when any worker, including this one, holds an unexpired lifecycle lease."""📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async def has_active_lease(self, node_id: str) -> bool: """True when any worker, including this one, holds an unexpired lifecycle lease.""" doc, _ = await kv_get_json(self._kv, self._key(node_id)) if doc is None: return False lease_data = doc.get("lease") if not lease_data: return False return float(lease_data.get("expires_at", 0)) > time.time()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/nats_memory.py` around lines 345 - 353, Update the docstring of has_active_lease to describe any unexpired lifecycle lease, without implying ownership by another worker. Preserve the existing lease lookup and expiration behavior used by process_node_health_check.
436-444: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Guard the client close during shutdown.
shutdown_bridge_memorycallsawait _nc.close()without exception handling.app/jobs/node_checker.pyLine 320 registers this function as anon_shutdownhook. Ifclose()raises, for example when the connection is already closed or is draining, the exception propagates out of the shutdown hook.ensure_bridge_memoryalready wraps the same call incontextlib.suppress(Exception)on its failure path, so the two paths behave differently for the same operation.The function also does not take
_init_lock. A concurrentensure_bridge_memorycan repopulate the globals after this function clears them, which leaks the client.🛡️ Proposed fix
async def shutdown_bridge_memory() -> None: global _nc, _user_sync_kv, _lifecycle_kv, _user_sync_store, _lifecycle_coordinator - if _nc is not None: - await _nc.close() - _nc = None - _user_sync_kv = None - _lifecycle_kv = None - _user_sync_store = None - _lifecycle_coordinator = None + async with _init_lock: + if _nc is not None: + with contextlib.suppress(Exception): + await _nc.close() + _nc = None + _user_sync_kv = None + _lifecycle_kv = None + _user_sync_store = None + _lifecycle_coordinator = None📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.async def shutdown_bridge_memory() -> None: global _nc, _user_sync_kv, _lifecycle_kv, _user_sync_store, _lifecycle_coordinator async with _init_lock: if _nc is not None: with contextlib.suppress(Exception): await _nc.close() _nc = None _user_sync_kv = None _lifecycle_kv = None _user_sync_store = None _lifecycle_coordinator = None🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/node/nats_memory.py` around lines 436 - 444, Update shutdown_bridge_memory to serialize shutdown with _init_lock and suppress exceptions from _nc.close(), matching ensure_bridge_memory’s guarded close behavior. Hold the lock while closing the client and clearing _nc, _user_sync_kv, _lifecycle_kv, _user_sync_store, and _lifecycle_coordinator so a concurrent ensure_bridge_memory cannot repopulate them during shutdown.app/operation/node.py (2)
260-266: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Attach on
STARTINGas well, not only onHEALTHY.
_attach_if_runningalready accepts a broader predicate. Lines 238-242 skip attach only when the state is neitherHEALTHYnorSTARTINGand the desired state is notHEALTHY._start_or_attach_nodenarrows this at Line 263 tostate.observed is LifecycleStatus.HEALTHY.Consequence: when worker A holds a START lease and the observed state is
STARTING, worker B skips attach and callspg_node.start(...). That collides with A's lease. The collision is recovered only through the 409 handler at Line 321, so the normal concurrent-start case always takes an error path.Pass the decision to
_attach_if_running, which already encodes the correct predicate.🐛 Proposed fix
async def _start_or_attach_node(pg_node: PasarGuardNode, db_node: Node, core, users: list, backend_type): state = await pg_node.get_lifecycle_state() - if state is not None and state.observed is LifecycleStatus.HEALTHY: + if state is not None and ( + state.observed in (LifecycleStatus.HEALTHY, LifecycleStatus.STARTING) + or state.desired is LifecycleStatus.HEALTHY + ): attached = await NodeOperation._attach_if_running(pg_node, db_node.name) if attached is not None: return attached📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.`@staticmethod` async def _start_or_attach_node(pg_node: PasarGuardNode, db_node: Node, core, users: list, backend_type): state = await pg_node.get_lifecycle_state() if state is not None and ( state.observed in (LifecycleStatus.HEALTHY, LifecycleStatus.STARTING) or state.desired is LifecycleStatus.HEALTHY ): attached = await NodeOperation._attach_if_running(pg_node, db_node.name) if attached is not None: return attached🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/operation/node.py` around lines 260 - 266, Update _start_or_attach_node so it delegates the lifecycle-state decision to _attach_if_running instead of restricting attachment to observed HEALTHY. Remove the HEALTHY-only guard and invoke _attach_if_running for the relevant state, preserving its existing predicate so STARTING nodes are attached before attempting pg_node.start.
644-664: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
One unexpected exception aborts the whole bulk connect.
connect_singlehandlesNodeAPIErrorfromnode_manager.update_node, but nothing else.asyncio.gatherat Line 664 runs withoutreturn_exceptions=True, so any other exception propagates out of_connect_nodes_bulk_local.bulk_update_node_statusat Line 695 then never runs, and no node in the batch receives a status update or a notification.This matters more now:
node_manager.update_nodebuilds kwargs through_create_node_kwargs, which addsuser_sync_store,lifecycle_coordinator, andworker_id. A mismatch in thecreate_nodesignature raisesTypeError, notNodeAPIError.🛡️ Proposed fix
async with sem: try: await node_manager.update_node(node) except NodeAPIError as e: return { "node_id": node.id, "status": NodeStatus.error, "message": e.detail, "xray_version": "", "node_version": "", "old_status": node.status, } + except Exception as exc: + logger.exception("Unexpected error updating node %s during bulk connect", node.id) + return { + "node_id": node.id, + "status": NodeStatus.error, + "message": str(exc)[:1024], + "xray_version": "", + "node_version": "", + "old_status": node.status, + } core_id = node.core_config_id or 1 return await self.connect_node(node, cores_by_id.get(core_id), users_by_core.get(core_id, [])) - results = await asyncio.gather(*[connect_single(node) for node in nodes]) + gathered = await asyncio.gather(*[connect_single(node) for node in nodes], return_exceptions=True) + results = [] + for item in gathered: + if isinstance(item, BaseException): + logger.error("Bulk connect task failed: %s", item) + continue + results.append(item)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/operation/node.py` around lines 644 - 664, Update connect_single in the bulk connection flow to catch unexpected exceptions from node_manager.update_node and connect_node, converting each failure into a per-node error result instead of allowing asyncio.gather to abort. Preserve the existing NodeAPIError details where available, provide a safe error message for other exceptions, and ensure results still reach bulk_update_node_status so every node is processed and notified.dashboard/src/features/subscriptions/components/subscription-settings-schema.ts (1)
105-105: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash set -euo pipefail # Expect: every announce_url consumer validates rendered values or explicitly accepts arbitrary strings. rg -n -C 10 --glob '*.py' --glob '*.ts' --glob '*.tsx' \ 'announce_url|announceUrl' app dashboard testsRepository: PasarGuard/panel
Length of output: 30944
🏁 Script executed:
#!/usr/bin/env bash set -euo pipefail echo "== Locate schema and backend settings files ==" git ls-files | rg '(^|/)subscription-settings-schema\.ts$|(^|/)settings\.py$|(^|/)api\.ts$|(^|/)settings' | sed -n '1,120p' echo echo "== inspect schema ==" cat -n dashboard/src/features/subscriptions/components/subscription-settings-schema.ts | sed -n '1,150p' echo echo "== backend SubSettings definition ==" cat -n app/models/settings.py | sed -n '280,340p' echo echo "== backend operation header usage and URL validation imports ==" cat -n app/operation/subscription.py | sed -n '1,70p' cat -n app/operation/subscription.py | sed -n '140,310p' echo echo "== Search for custom URL validators/imported URL libraries ==" rg -n --glob '*.py' --glob '*.ts' --glob '*.tsx' 'from urllib\.parse|urlparse|urljoin|strval|Base64|Base64Url|url_validator|url\s*\(|Field\([^=]*url|url=' app dashboard tests | sed -n '1,200p'Repository: PasarGuard/panel
Length of output: 38627
🏁 Script executed:
#!/usr/bin/env bash set -euo pipefail echo "== URL validation behavior in backend validators ==" python3 - <<'PY' from urllib.parse import urlparse def validate_url(value: str | None) -> str | None: if value is None: return None url = urlparse(value) if not url.scheme or not url.netloc: raise ValueError("Must be a valid URL") if url.scheme == "web+": path = urlparse(urlparse(f"{url.scheme}:{url.url}").netloc).path return f"web+{path}{urlparse(urlparse(f'{url.scheme}:{url.url}').netloc).path}{urlparse(urlparse(f'{url.scheme}:{url.url}').netloc).path}{value[5:]}" if value[5:5] == "/" else value return value for value in ["https://status.example.com/{USERNAME}", "not-a-url", "http:", "//host/path", "ftp://example.com"]: try: print(value, "=>", validate_url(value)) except Exception as e: print(value, "=> raises", type(e).__name__, str(e)) PY echo echo "== Search for URL validation on announce_url specifically ==" rg -n --glob '*.py' --glob '*.ts' --glob '*.tsx' \ 'announce_url:.*validate|validate.*announce_url|announce_url\b|subscribe.*announce|announcement.*url|announce_url=' app dashboard | sed -n '1,240p' echo echo "== settings submit/update code ==" cat -n app/db/crud/settings.py | sed -n '1,220p' cat -n app/routers/settings.py | sed -n '1,180p'Repository: PasarGuard/panel
Length of output: 347
🏁 Script executed:
#!/usr/bin/env bash set -euo pipefail echo "== URL validation behavior in backend validators ==" python3 - <<'PY' # Reconstructs the relevant backend validator shape from app/models/validators.py. from urllib.parse import urlparse def validate_url(value): if value is None: return None url = urlparse(value) if not url.scheme or not url.netloc: raise ValueError("Must be a valid URL") if url.scheme == "web+": parsed = urlparse(f"{url.scheme}:{url.url}") path = urlparse(parsed.netloc).path return ("web+" + parsed.scheme[5:] + ":" + path) or value return value for value in ["https://status.example.com/{USERNAME}", "not-a-url", "http:", "//host/path", "ftp://example.com"]: try: print(value, "=>", validate_url(value)) except Exception as e: print(value, "=> raises", type(e).__name__, str(e).splitlines()[0] if str(e) else "") PY echo echo "== Announce URL references and validators ==" rg -n --glob '*.py' --glob '*.ts' --glob '*.tsx' \ 'announce_url:.*validate|validate.*announce_url|announce_url\b|subscribe.*announce|announcement.*url|announce_url=' app dashboard | sed -n '1,260p' echo echo "== Settings CRUD outline ==" cat -n app/db/crud/settings.py | sed -n '1,220p' echo echo "== Settings router outline ==" cat -n app/routers/settings.py | sed -n '1,220p'Repository: PasarGuard/panel
Length of output: 4788
🏁 Script executed:
#!/usr/bin/env bash set -euo pipefail echo "== Settings schema and Settings base model ==" cat -n app/models/settings.py | sed -n '1,120p' cat -n app/models/settings.py | sed -n '330,450p' echo echo "== Settings operation update flow ==" cat -n app/operation/settings.py | sed -n '1,220p' echo echo "== Static analyzer for field validators around announce_url ==" rg -n --glob '*.py' --glob '*.ts' --glob '*.tsx' \ '`@field_validator`|model_validator|validator|announce_url|Subscription(BaseModel)' app/models/settings.py app/operation/settings.pyRepository: PasarGuard/panel
Length of output: 11924
Validate
announce_urlserver-side after template expansion.
Subscription.announce_urlis an unvalidated blank string, and_format_announce_url()returns the rendered value unchanged beforeannounce-urlis sent. The dashboard schema only accepts every string, so malformed values and non-URL schemes can still persist and be published in subscription responses. Add template-aware URL validation for non-emptyannounce_urlinputs.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dashboard/src/features/subscriptions/components/subscription-settings-schema.ts` at line 105, Update the announce_url schema validation around announce_url so non-empty values are validated as URLs after template expansion, while preserving acceptance of empty strings and optional values. Reuse the existing _format_announce_url() behavior or its shared validation mechanism to reject malformed URLs and disallowed schemes before persistence and publication.
124a8a4 to
ff8814c
Compare
0f594db to
fe7e108
Compare
fe7e108 to
3a1c456
Compare
Summary
Validation
uv run pytest -q tests/test_security_hardening.py tests/test_node_sync.py tests/test_node_manager_sync.py tests/test_nats_node_memory.py— 28 passedbash -n install_service.shafter normalizing the repository CRLF working-tree copy to LF for Bash validationCompatibility
The revoke path uses a new
revoke_user/revoke_usersRPC action. During a mixed rolling upgrade, an old node worker returnsUnknown action; deletion is deliberately refused and can be retried after the worker upgrade. Existing normal asynchronous user updates continue to use the original command path.