Add staff visibility workstation enrollment registration vault and category task views

This commit is contained in:
A R R R Associates
2026-09-02 14:05:20 +05:30
parent 1fe1597d6f
commit 97777e13a1
22 changed files with 538 additions and 103 deletions
+3 -2
View File
@@ -4,13 +4,13 @@ import io
from pathlib import Path
import zipfile
ERP_LOCAL_AGENT_VERSION = "1.21.1"
ERP_LOCAL_AGENT_VERSION = "1.22.0"
ERP_LOCAL_AGENT_NAME = "ERP Local Agent"
RUNTIME_ROOT = Path(__file__).resolve().parent / "local_agent_runtime"
_DETERMINISTIC_ZIP_TIMESTAMP = (2026, 1, 1, 0, 0, 0)
def build_agent_env(*, erp_base_url: str, node_code: str, node_secret: str, storage_root: str, tenant_id=None, branch_id=None, sync_interval_seconds: int = 30, request_timeout_seconds: int = 60, tunnel_enabled: bool = True, tunnel_reconnect_seconds: int = 10) -> str:
def build_agent_env(*, erp_base_url: str, node_code: str, node_secret: str, storage_root: str, tenant_id=None, branch_id=None, enrollment_token: str | None = None, sync_interval_seconds: int = 30, request_timeout_seconds: int = 60, tunnel_enabled: bool = True, tunnel_reconnect_seconds: int = 10) -> str:
erp_base_url = (erp_base_url or "").strip().rstrip("/")
if erp_base_url.startswith("http://"):
host = erp_base_url[7:].split("/", 1)[0].split(":", 1)[0].lower()
@@ -19,6 +19,7 @@ def build_agent_env(*, erp_base_url: str, node_code: str, node_secret: str, stor
storage_root = (storage_root or r"D:\AuditFirmStorage").strip()
return (
f"ERP_BASE_URL={erp_base_url}\n" f"NODE_CODE={(node_code or '').strip()}\n" f"NODE_SECRET={(node_secret or '').strip()}\n"
f"ENROLLMENT_TOKEN={(enrollment_token or '').strip()}\n"
f"STORAGE_ROOT={storage_root}\n" f"TENANT_ID={'' if tenant_id is None else tenant_id}\n" f"AUDIT_FIRM_ID={'' if tenant_id is None else tenant_id}\n"
f"BRANCH_ID={'' if branch_id is None else branch_id}\n" f"SYNC_INTERVAL_SECONDS={int(sync_interval_seconds or 30)}\n" f"POLL_INTERVAL_SECONDS={int(sync_interval_seconds or 30)}\n"
f"REQUEST_TIMEOUT_SECONDS={int(request_timeout_seconds or 60)}\n" f"TUNNEL_ENABLED={str(bool(tunnel_enabled)).lower()}\n" f"TUNNEL_RECONNECT_SECONDS={int(tunnel_reconnect_seconds or 10)}\n"
@@ -1,2 +1,2 @@
__version__ = "1.21.1"
__version__ = "1.22.0"
AGENT_NAME = "ERP Local Agent"
@@ -1,96 +1,114 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Iterable
from typing import Any
import requests
from .config import AgentConfig
class ERPClient:
def __init__(self, config: AgentConfig):
AUTH_FILE = "workstation_auth.json"
def __init__(self, config: AgentConfig, *, workstation: dict[str, Any] | None = None, root: Path | None = None):
self.config = config
self.workstation = workstation or {}
self.root = root or Path.cwd()
self.session = requests.Session()
self.session.headers.update(config.headers)
self.node_secret = self._resolve_node_secret()
self.session.headers.update({
"X-Node-Code": self.config.node_code,
"X-Node-Secret": self.node_secret,
"User-Agent": self.config.headers.get("User-Agent", "ERPLocalAgent"),
})
instance_id = str(self.workstation.get("agent_instance_id") or self.config.agent_instance_id or "").strip()
if instance_id:
self.session.headers["X-Agent-Instance-ID"] = instance_id
def _url(self, path: str) -> str:
return f"{self.config.erp_base_url}{path}"
def heartbeat(self, payload: dict[str, Any]) -> dict[str, Any]:
response = self.session.post(
self._url("/documents/storage-agent/heartbeat"),
json=payload,
timeout=self.config.request_timeout_seconds,
def _auth_path(self) -> Path:
path = self.root / "data" / self.AUTH_FILE
path.parent.mkdir(parents=True, exist_ok=True)
return path
def _cached_workstation_secret(self) -> str:
path = self._auth_path()
try:
payload = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
except Exception:
return ""
if str(payload.get("node_code") or "") != self.config.node_code:
return ""
expected_instance = str(self.workstation.get("agent_instance_id") or "")
if expected_instance and str(payload.get("agent_instance_id") or "") != expected_instance:
return ""
return str(payload.get("workstation_secret") or "").strip()
def _resolve_node_secret(self) -> str:
if self.config.node_secret:
return self.config.node_secret
cached = self._cached_workstation_secret()
if cached:
return cached
token = (self.config.enrollment_token or "").strip()
if not token:
raise RuntimeError("Local Agent has neither a node secret nor a workstation enrollment token.")
response = requests.post(
self._url("/documents/storage-agent/enroll-workstation"),
json={"enrollment_token": token, "workstation": self.workstation},
timeout=max(30, self.config.request_timeout_seconds),
headers={"User-Agent": self.config.headers.get("User-Agent", "ERPLocalAgent")},
)
response.raise_for_status()
return response.json() if response.content else {"status": "ok"}
data = response.json()
secret = str(data.get("workstation_secret") or "").strip()
if not secret:
raise RuntimeError(data.get("error") or "ERP did not return a workstation credential.")
path = self._auth_path()
payload = {
"node_code": self.config.node_code,
"agent_instance_id": str(self.workstation.get("agent_instance_id") or ""),
"workstation_secret": secret,
}
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
tmp.replace(path)
return secret
def heartbeat(self, payload: dict[str, Any]) -> dict[str, Any]:
response = self.session.post(self._url("/documents/storage-agent/heartbeat"), json=payload, timeout=self.config.request_timeout_seconds)
response.raise_for_status(); return response.json() if response.content else {"status": "ok"}
def pending_storage_jobs(self) -> list[dict[str, Any]]:
response = self.session.get(
self._url("/documents/storage-agent/jobs/pending"),
timeout=self.config.request_timeout_seconds,
)
response.raise_for_status()
data = response.json()
if isinstance(data, list):
return data
return data.get("jobs", [])
response = self.session.get(self._url("/documents/storage-agent/jobs/pending"), timeout=self.config.request_timeout_seconds)
response.raise_for_status(); data = response.json(); return data if isinstance(data, list) else data.get("jobs", [])
def download_storage_job(self, job_id: int | str):
response = self.session.get(
self._url(f"/documents/storage-agent/jobs/{job_id}/download"),
stream=True,
timeout=self.config.request_timeout_seconds,
)
response.raise_for_status()
return response
response = self.session.get(self._url(f"/documents/storage-agent/jobs/{job_id}/download"), stream=True, timeout=self.config.request_timeout_seconds)
response.raise_for_status(); return response
def acknowledge_storage_job(self, job_id: int | str, payload: dict[str, Any]) -> dict[str, Any]:
response = self.session.post(
self._url(f"/documents/storage-agent/jobs/{job_id}/ack"),
json=payload,
timeout=self.config.request_timeout_seconds,
)
response.raise_for_status()
return response.json() if response.content else {"status": "ok"}
response = self.session.post(self._url(f"/documents/storage-agent/jobs/{job_id}/ack"), json=payload, timeout=self.config.request_timeout_seconds)
response.raise_for_status(); return response.json() if response.content else {"status": "ok"}
def pending_download_requests(self) -> list[dict[str, Any]]:
response = self.session.get(
self._url("/documents/storage-agent/download-requests/pending"),
timeout=self.config.request_timeout_seconds,
)
response.raise_for_status()
data = response.json()
if isinstance(data, list):
return data
return data.get("requests", [])
response = self.session.get(self._url("/documents/storage-agent/download-requests/pending"), timeout=self.config.request_timeout_seconds)
response.raise_for_status(); data = response.json(); return data if isinstance(data, list) else data.get("requests", [])
def upload_download_request_file(self, request_id: int | str, file_path: Path, extra: dict[str, Any]) -> dict[str, Any]:
with file_path.open("rb") as handle:
files = {"file": (file_path.name, handle, "application/octet-stream")}
data = {key: str(value) for key, value in extra.items() if value is not None}
response = self.session.post(
self._url(f"/documents/storage-agent/download-requests/{request_id}/upload"),
files=files,
data=data,
timeout=max(self.config.request_timeout_seconds, 300),
)
response.raise_for_status()
return response.json() if response.content else {"status": "ok"}
response = self.session.post(self._url(f"/documents/storage-agent/download-requests/{request_id}/upload"), files=files, data=data, timeout=max(self.config.request_timeout_seconds, 300))
response.raise_for_status(); return response.json() if response.content else {"status": "ok"}
def update_manifest(self) -> dict[str, Any]:
response = self.session.get(
self._url("/documents/erp-local-agent/update-manifest"),
timeout=self.config.request_timeout_seconds,
)
response.raise_for_status()
return response.json()
response = self.session.get(self._url("/documents/erp-local-agent/update-manifest"), timeout=self.config.request_timeout_seconds)
response.raise_for_status(); return response.json()
def download_update_package(self) -> bytes:
response = self.session.get(
self._url("/documents/erp-local-agent/update-package"),
timeout=max(self.config.request_timeout_seconds, 300),
)
response.raise_for_status()
return response.content
response = self.session.get(self._url("/documents/erp-local-agent/update-package"), timeout=max(self.config.request_timeout_seconds, 300))
response.raise_for_status(); return response.content
@@ -16,6 +16,8 @@ class AgentConfig:
node_code: str
node_secret: str
storage_root: Path
enrollment_token: str | None = None
agent_instance_id: str | None = None
tenant_id: str | None = None
branch_id: str | None = None
poll_interval_seconds: int = 30
@@ -34,11 +36,14 @@ class AgentConfig:
@property
def headers(self) -> dict[str, str]:
return {
headers = {
"X-Node-Code": self.node_code,
"X-Node-Secret": self.node_secret,
"User-Agent": f"ERPLocalAgent/{__version__}",
}
if self.agent_instance_id:
headers["X-Agent-Instance-ID"] = self.agent_instance_id
return headers
@property
def tunnel_url(self) -> str:
@@ -49,7 +54,10 @@ class AgentConfig:
ws_base = "ws://" + base[len("http://"):]
else:
ws_base = base
query = urlencode({"node_code": self.node_code, "node_secret": self.node_secret})
query_data = {"node_code": self.node_code, "node_secret": self.node_secret}
if self.agent_instance_id:
query_data["agent_instance_id"] = self.agent_instance_id
query = urlencode(query_data)
return f"{ws_base}/documents/storage-agent/tunnel?{query}"
@@ -79,6 +87,7 @@ def load_config(env_file: str | None = None) -> AgentConfig:
erp_base_url = os.getenv("ERP_BASE_URL", "").rstrip("/")
node_code = os.getenv("NODE_CODE", "").strip()
node_secret = os.getenv("NODE_SECRET", "").strip()
enrollment_token = os.getenv("ENROLLMENT_TOKEN", "").strip() or None
storage_root_raw = os.getenv("STORAGE_ROOT", "").strip()
tenant_id = os.getenv("TENANT_ID", os.getenv("AUDIT_FIRM_ID", "")).strip() or None
branch_id = os.getenv("BRANCH_ID", "").strip() or None
@@ -88,8 +97,8 @@ def load_config(env_file: str | None = None) -> AgentConfig:
missing.append("ERP_BASE_URL")
if not node_code:
missing.append("NODE_CODE")
if not node_secret:
missing.append("NODE_SECRET")
if not node_secret and not enrollment_token:
missing.append("NODE_SECRET or ENROLLMENT_TOKEN")
if not storage_root_raw:
missing.append("STORAGE_ROOT")
if missing:
@@ -103,6 +112,7 @@ def load_config(env_file: str | None = None) -> AgentConfig:
node_code=node_code,
node_secret=node_secret,
storage_root=storage_root,
enrollment_token=enrollment_token,
tenant_id=tenant_id,
branch_id=branch_id,
poll_interval_seconds=_get_int("POLL_INTERVAL_SECONDS", 30),
@@ -2,6 +2,7 @@ from __future__ import annotations
import argparse
import asyncio
from dataclasses import replace
from pathlib import Path
import threading
import time
@@ -42,7 +43,12 @@ def main() -> int:
db.set_meta("machine_name", workstation["machine_name"])
db.set_meta("machine_fingerprint", workstation["machine_fingerprint"])
db.record_event("INFO", "agent_started", f"ERP Local Agent {__version__} started")
client = ERPClient(config)
client = ERPClient(config, workstation=workstation, root=root)
config = replace(
config,
node_secret=client.node_secret,
agent_instance_id=workstation["agent_instance_id"],
)
agent = StorageAgent(config, client, db, logger)
updater = AgentUpdater(config, client, logger, root, db=db)
dashboard = AgentDashboard(config, db, updater, logger, root)
+1
View File
@@ -205,6 +205,7 @@ class ERPWorkstationAgent(CommonBase):
machine_name: Mapped[str] = mapped_column(String(200), nullable=False)
platform_name: Mapped[str | None] = mapped_column(String(120), nullable=True)
agent_version: Mapped[str | None] = mapped_column(String(40), nullable=True, index=True)
workstation_secret_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
capabilities_json: Mapped[str | None] = mapped_column(Text, nullable=True)
tally_connected: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True)
tally_company_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+48
View File
@@ -25,6 +25,7 @@ from app.modules.documents.models import (
DocumentStorageJob,
EngagementDocument,
EngagementDocumentVersion,
ERPWorkstationAgent,
PermanentClientDocument,
PermanentClientDocumentVersion,
PermanentDocumentDownloadRequest,
@@ -813,6 +814,53 @@ def authenticate_storage_node(db: Session, node_code: str | None, secret: str |
return node
def authenticate_storage_agent(
db: Session,
node_code: str | None,
secret: str | None,
agent_instance_id: str | None = None,
request=None,
) -> BranchStorageNode | None:
"""Authenticate either a legacy branch-node secret or a workstation-specific secret.
Existing Local Agents continue to use the branch node secret unchanged. New
workstation enrollment packages receive a workstation-only secret so adding a
PC no longer rotates credentials used by already-installed office systems.
"""
node = authenticate_storage_node(db, node_code, secret, request=request)
if node:
return node
instance_id = (agent_instance_id or "").strip()
if not node_code or not secret or not instance_id:
return None
node = db.execute(
select(BranchStorageNode).where(
BranchStorageNode.node_code == node_code,
BranchStorageNode.is_active.is_(True),
)
).scalar_one_or_none()
if not node:
return None
workstation = db.execute(
select(ERPWorkstationAgent).where(
ERPWorkstationAgent.storage_node_id == node.id,
ERPWorkstationAgent.agent_instance_id == instance_id,
ERPWorkstationAgent.is_active.is_(True),
)
).scalar_one_or_none()
if not workstation or not workstation.workstation_secret_hash:
return None
if not hmac.compare_digest(workstation.workstation_secret_hash, hash_storage_secret(secret)):
return None
now = datetime.now(timezone.utc)
node.last_seen_at_utc = now
workstation.last_seen_at_utc = now
ip = request.client.host if request and request.client else None
node.last_seen_ip = ip
workstation.last_seen_ip = ip
return node
def list_pending_storage_jobs(db: Session, node: BranchStorageNode, limit: int = 20):
return db.execute(
select(DocumentStorageJob)
@@ -149,10 +149,16 @@
{% if node.status == 'disabled_duplicate' %}
<span class="text-xs text-slate-500">Duplicate disabled</span>
{% else %}
<form method="post" action="/documents/storage-nodes/{{ node.id }}/download-new-workstation">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<input type="hidden" name="storage_root_path" value="{{ node.storage_root_path or 'D:\AuditFirmStorage' }}">
<button class="rounded-lg bg-emerald-600 px-3 py-1.5 text-xs font-semibold text-white">+ New Workstation</button>
</form>
<div class="max-w-48 text-right text-[11px] text-slate-500">Installs on another office PC without changing credentials used by existing workstations.</div>
<form method="post" action="/documents/storage-nodes/{{ node.id }}/download-agent-package">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<input type="hidden" name="storage_root_path" value="{{ node.storage_root_path or 'D:\AuditFirmStorage' }}">
<button class="text-sm text-emerald-700 font-semibold">Download ERP Local Agent</button>
<button class="text-xs font-semibold text-amber-700">Reissue branch package</button>
</form>
<form method="post" action="/documents/storage-nodes/{{ node.id }}/toggle">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
@@ -173,7 +179,7 @@
<div class="rounded-2xl border bg-white overflow-hidden shadow-sm">
<div class="p-4 border-b">
<h2 class="font-bold text-slate-900">Registered Workstations</h2>
<p class="text-xs text-slate-500 mt-1">Each PC keeps a persistent agent identity. The branch storage node remains unchanged; multiple Tally workstations can report through the same outbound tunnel credentials.</p>
<p class="text-xs text-slate-500 mt-1">Each PC keeps a persistent agent identity. Use <strong>+ New Workstation</strong> on the branch node above for another office PC; it receives its own workstation credential and existing agents keep working.</p>
</div>
<div class="overflow-x-auto">
<table class="w-full text-sm">
+125 -3
View File
@@ -17,6 +17,7 @@ from sqlalchemy.orm import joinedload
from app.core.db.common import CommonSessionLocal
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
from app.core.security.session_auth import get_current_user
from app.core.security.jwt_tokens import decode_token, encode_access_token
from app.core.templating import templates
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.core.rbac.permission_guard import require_permission
@@ -33,6 +34,7 @@ from app.modules.documents.services import (
cleanup_completed_permanent_storage_job_vps_stage,
retry_verified_vps_cleanup_for_node,
authenticate_storage_node,
authenticate_storage_agent,
create_branch_storage_node,
generate_storage_secret,
hash_storage_secret,
@@ -1578,6 +1580,122 @@ def download_storage_node_env(request: Request, csrf_token: str = Form(...)):
return RedirectResponse(url="/documents/storage-nodes?error=env_download_disabled", status_code=303)
@router.post("/storage-nodes/{node_id}/download-new-workstation")
def download_new_workstation_agent(
request: Request,
node_id: int,
storage_root_path: str = Form(r"D:\AuditFirmStorage"),
csrf_token: str = Form(...),
):
"""Download a short-lived enrollment package without rotating existing agent credentials."""
validate_csrf(request, csrf_token)
db = CommonSessionLocal()
try:
user, response = _require_user(request, db, "documents.upload")
if response:
return response
scope = build_document_scope(request, db, user)
if not _can_manage_branch_storage(scope):
return _redirect_denied()
node = db.get(BranchStorageNode, node_id)
if not _node_allowed_for_storage_scope(node, user, scope):
return _redirect_denied()
token = encode_access_token(
{
"purpose": "workstation_enrollment",
"node_id": int(node.id),
"node_code": node.node_code,
"tenant_id": int(node.tenant_id),
"branch_id": int(node.branch_id) if node.branch_id is not None else None,
},
expires_minutes=30,
)
env_text = build_agent_env(
erp_base_url=str(request.base_url).rstrip("/"),
node_code=node.node_code,
node_secret="",
enrollment_token=token,
storage_root=_effective_storage_root(node, storage_root_path),
tenant_id=node.tenant_id,
branch_id=node.branch_id,
)
package = build_preconfigured_agent_zip(
env_text=env_text, include_admin_readme=bool(scope.is_system_admin)
)
filename = _agent_download_filename(f"{node.node_code}-NEW-WORKSTATION", ".zip")
return Response(
package,
media_type="application/zip",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
finally:
db.close()
@router.post("/storage-agent/enroll-workstation")
async def enroll_workstation(request: Request):
"""Exchange a short-lived enrollment token for a workstation-only agent secret."""
payload = await request.json()
token = str(payload.get("enrollment_token") or "").strip()
workstation_payload = payload.get("workstation") if isinstance(payload.get("workstation"), dict) else {}
try:
claims = decode_token(token)
except Exception:
return JSONResponse({"ok": False, "error": "invalid_or_expired_enrollment_token"}, status_code=401)
if claims.get("purpose") != "workstation_enrollment":
return JSONResponse({"ok": False, "error": "invalid_enrollment_token_type"}, status_code=401)
instance_id = str(workstation_payload.get("agent_instance_id") or "").strip()
fingerprint = str(workstation_payload.get("machine_fingerprint") or "").strip()
machine_name = str(workstation_payload.get("machine_name") or "").strip()
if not instance_id or not fingerprint or not machine_name:
return JSONResponse({"ok": False, "error": "workstation_identity_required"}, status_code=400)
db = CommonSessionLocal()
try:
node = db.get(BranchStorageNode, int(claims.get("node_id") or 0))
if (
not node
or not node.is_active
or node.node_code != claims.get("node_code")
or int(node.tenant_id) != int(claims.get("tenant_id") or 0)
):
return JSONResponse({"ok": False, "error": "enrollment_node_unavailable"}, status_code=403)
row = db.execute(
select(ERPWorkstationAgent).where(
ERPWorkstationAgent.storage_node_id == node.id,
ERPWorkstationAgent.agent_instance_id == instance_id,
)
).scalar_one_or_none()
if row is None:
row = ERPWorkstationAgent(
tenant_id=node.tenant_id,
branch_id=node.branch_id,
storage_node_id=node.id,
agent_instance_id=instance_id,
machine_fingerprint=fingerprint,
machine_name=machine_name[:200],
)
db.add(row)
secret = generate_storage_secret()
row.workstation_secret_hash = hash_storage_secret(secret)
row.machine_fingerprint = fingerprint
row.machine_name = machine_name[:200]
row.platform_name = str(workstation_payload.get("platform_name") or "")[:120] or None
row.agent_version = str(workstation_payload.get("agent_version") or ERP_LOCAL_AGENT_VERSION)[:40] or None
row.is_active = True
row.status = "enrolled"
row.last_seen_at_utc = datetime.now(timezone.utc)
row.last_seen_ip = request.client.host if request.client else None
db.commit()
return {
"ok": True,
"node_code": node.node_code,
"agent_instance_id": instance_id,
"workstation_secret": secret,
}
finally:
db.close()
@router.post("/storage-nodes/download-agent-package")
def download_preconfigured_storage_agent(
request: Request,
@@ -1694,7 +1812,10 @@ def storage_jobs(request: Request, status: str = ""):
def _agent_auth(db, request: Request, x_node_code: str | None, x_node_secret: str | None):
node = authenticate_storage_node(db, x_node_code, x_node_secret, request=request)
instance_id = (request.headers.get("x-agent-instance-id") or "").strip()
node = authenticate_storage_agent(
db, x_node_code, x_node_secret, agent_instance_id=instance_id, request=request
)
if not node:
return None, JSONResponse({"ok": False, "error": "invalid_storage_node_credentials"}, status_code=401)
return node, None
@@ -1942,11 +2063,12 @@ async def storage_agent_tunnel(websocket: WebSocket):
"""
node_code = websocket.query_params.get("node_code") or websocket.headers.get("x-node-code")
node_secret = websocket.query_params.get("node_secret") or websocket.headers.get("x-node-secret")
agent_instance_id = websocket.query_params.get("agent_instance_id") or websocket.headers.get("x-agent-instance-id")
await websocket.accept()
db = CommonSessionLocal()
try:
node = authenticate_storage_node(db, node_code, node_secret, request=None)
node = authenticate_storage_agent(db, node_code, node_secret, agent_instance_id=agent_instance_id, request=None)
if not node:
await websocket.send_json({"ok": False, "type": "error", "error": "invalid_storage_node_credentials"})
await websocket.close(code=1008)
@@ -1971,7 +2093,7 @@ async def storage_agent_tunnel(websocket: WebSocket):
db = CommonSessionLocal()
try:
node = authenticate_storage_node(db, node_code, node_secret, request=None)
node = authenticate_storage_agent(db, node_code, node_secret, agent_instance_id=agent_instance_id, request=None)
if not node:
await websocket.send_json({"ok": False, "type": "error", "error": "node_deactivated_or_invalid"})
await websocket.close(code=1008)