Nexus/verify.py

#!/usr/bin/env python3
"""Local verification script — validates modules, routes, and security"""
import sys
sys.path.insert(0, '.')

# 1. 验证所有API模块导入
print('=== 1. 模块导入验证 ===')
modules = [
    'server.api.install',
    'server.api.servers',
    'server.api.auth',
    'server.api.auth_jwt',
    'server.api.agent',
    'server.api.scripts',
    'server.api.settings',
    'server.api.assets',
    'server.api.sync_v2',
    'server.api.websocket',
    'server.api.webssh',
    'server.api.health',
    'server.api.dependencies',
    'server.api.schemas',
]
import_errors = []
for m in modules:
    try:
        __import__(m)
        print(f'  OK {m}')
    except Exception as e:
        print(f'  FAIL {m}: {e}')
        import_errors.append(m)

# 2. 验证domain模型
print('\n=== 2. Domain模型验证 ===')
from server.domain.models import Base
tables = sorted(Base.metadata.tables.keys())
print(f'  表数量: {len(tables)}')
for t in tables:
    print(f'  OK {t}')

# 3. 验证config
print('\n=== 3. Config验证 ===')
from server.config import settings
sk = "(empty)" if not settings.SECRET_KEY else "(set)"
print(f'  SYSTEM_NAME: {settings.SYSTEM_NAME}')
print(f'  PORT: {settings.PORT}')
print(f'  DB_POOL_SIZE: {settings.DB_POOL_SIZE}')
print(f'  SECRET_KEY: {sk}')

# 4. 验证FastAPI app
print('\n=== 4. FastAPI App验证 ===')
from server.main import app, INSTALL_MODE
print(f'  INSTALL_MODE: {INSTALL_MODE}')
routes = [r for r in app.routes if hasattr(r, 'path')]
print(f'  总路由数: {len(routes)}')

# 5. 验证关键API路由
print('\n=== 5. API路由验证 ===')
api_routes = sorted(set(r.path for r in routes if hasattr(r, 'path') and r.path.startswith('/api')))
for r in api_routes:
    methods = set()
    for route in routes:
        if hasattr(route, 'path') and route.path == r and hasattr(route, 'methods'):
            methods.update(route.methods)
    mstr = " ".join(sorted(methods))
    print(f'  {mstr:20s} {r}')

# 6. 验证install API
print('\n=== 6. Install API验证 ===')
from server.api.install import router as install_router
for r in install_router.routes:
    if hasattr(r, 'path'):
        methods = getattr(r, 'methods', set())
        mstr = " ".join(sorted(methods))
        print(f'  {mstr:20s} {r.path}')

# 7. 验证JWT保护 — 检查关键路由是否依赖get_current_admin
print('\n=== 7. JWT保护验证 ===')
from server.api.auth_jwt import get_current_admin
jwt_protected = []
unprotected = []
for r in routes:
    if not hasattr(r, 'path') or not r.path.startswith('/api'):
        continue
    if not hasattr(r, 'endpoint'):
        continue
    # Check if get_current_admin is in the dependency list
    deps = getattr(r, 'dependant', None)
    has_jwt = False
    if deps:
        for dep in (deps.dependencies or []):
            if dep.call is get_current_admin or (hasattr(dep.call, '__name__') and dep.call.__name__ == 'get_current_admin'):
                has_jwt = True
                break
    if has_jwt:
        jwt_protected.append(r.path)
    else:
        unprotected.append(r.path)

print('  JWT保护的路由:')
for r in sorted(set(jwt_protected)):
    print(f'    + {r}')
print('  无JWT的路由(应仅限install/agent/health/login/refresh):')
for r in sorted(set(unprotected)):
    print(f'    - {r}')

# 8. 验证Settings安全
print('\n=== 8. Settings安全验证 ===')
from server.api.settings import IMMUTABLE_KEYS, SENSITIVE_KEYS
print(f'  不可改项: {IMMUTABLE_KEYS}')
print(f'  敏感字段: {SENSITIVE_KEYS}')

# 9. 验证后台任务模块
print('\n=== 9. 后台任务验证 ===')
bg_modules = [
    'server.background.heartbeat_flush',
    'server.background.self_monitor',
    'server.background.schedule_runner',
    'server.background.retry_runner',
]
for m in bg_modules:
    try:
        __import__(m)
        print(f'  OK {m}')
    except Exception as e:
        print(f'  FAIL {m}: {e}')

# 10. 验证基础设施模块
print('\n=== 10. 基础设施验证 ===')
infra_modules = [
    'server.infrastructure.database.crypto',
    'server.infrastructure.database.session',
    'server.infrastructure.redis.client',
    'server.infrastructure.ssh.asyncssh_pool',
    'server.infrastructure.telegram',
]
for m in infra_modules:
    try:
        __import__(m)
        print(f'  OK {m}')
    except Exception as e:
        print(f'  FAIL {m}: {e}')

# Summary
print('\n' + '=' * 50)
if import_errors:
    print(f'FAILED: {len(import_errors)} module(s) failed to import')
    sys.exit(1)
else:
    print('ALL CHECKS PASSED')