Nexus/audit_scan.py

"""Full codebase audit scanner"""
import sys, os, re
sys.path.insert(0, '.')

# === 1. Backend API endpoints ===
print('=' * 60)
print('1. BACKEND ROUTE REGISTRY')
print('=' * 60)

api_files = [f for f in os.listdir('server/api') if f.endswith('.py') and f not in ('__init__.py', 'dependencies.py', 'schemas.py')]
for f in sorted(api_files):
    try:
        with open(f'server/api/{f}', 'r', encoding='utf-8') as fh:
            content = fh.read()
        prefix_match = re.search(r'prefix=["\'](/api/[^"\']+)["\']', content)
        prefix = prefix_match.group(1) if prefix_match else '???'
        routes = re.findall(r'@router\.(get|post|put|delete|websocket)\(["\'](.*?)["\']', content)
        if routes:
            print(f'  {f}: prefix={prefix}')
            for method, path in routes:
                print(f'    {method.upper():4s} {prefix}{path}')
    except Exception as e:
        print(f'  {f}: ERROR {e}')

# === 2. Frontend API calls ===
print()
print('=' * 60)
print('2. FRONTEND API CALLS')
print('=' * 60)

html_files = [f for f in os.listdir('web/app') if f.endswith('.html')]
for f in sorted(html_files):
    try:
        with open(f'web/app/{f}', 'r', encoding='utf-8') as fh:
            content = fh.read()
        pattern = r"API\s*\+\s*['\"](/[^'\"]+)['\"]"
        calls = re.findall(pattern, content)
        if calls:
            print(f'  {f}:')
            for call in sorted(set(calls)):
                print(f'    {call}')
    except Exception as e:
        print(f'  {f}: ERROR {e}')

# === 3. Missing imports (AuditLog used but not imported) ===
print()
print('=' * 60)
print('3. POTENTIAL MISSING IMPORTS (AuditLog used without import)')
print('=' * 60)

for f in sorted(api_files):
    try:
        with open(f'server/api/{f}', 'r', encoding='utf-8') as fh:
            content = fh.read()
        uses_audit = 'AuditLog(' in content
        imports_audit = 'from server.domain.models import' in content and 'AuditLog' in content.split('from server.domain.models import')[1].split('\n')[0] if 'from server.domain.models import' in content else False
        # Also check local imports
        local_imports_audit = 'from server.domain.models import AuditLog' in content or ('AuditLog' in content and 'import AuditLog' in content)
        if uses_audit and not imports_audit and not local_imports_audit:
            # Check if it's imported via local import inside function
            has_local = 'import AuditLog' in content or 'from server.domain.models import' in content
            if not has_local:
                print(f'  [BUG] {f}: Uses AuditLog() but never imported!')
            else:
                print(f'  [WARN] {f}: Uses AuditLog() via local import only')
    except Exception as e:
        pass

# === 4. SSH pool usage (acquire without release) ===
print()
print('=' * 60)
print('4. SSH POOL: acquire() without release()')
print('=' * 60)

for root, dirs, files in os.walk('server'):
    for f in files:
        if not f.endswith('.py'):
            continue
        path = os.path.join(root, f)
        try:
            with open(path, 'r', encoding='utf-8') as fh:
                content = fh.read()
            acquires = len(re.findall(r'ssh_pool\.acquire', content))
            releases = len(re.findall(r'ssh_pool\.release', content))
            if acquires > 0 and acquires != releases:
                print(f'  [WARN] {path}: acquire={acquires} release={releases}')
        except:
            pass

# === 5. AsyncSessionLocal without commit ===
print()
print('=' * 60)
print('5. DB writes without commit()')
print('=' * 60)

for root, dirs, files in os.walk('server'):
    for f in files:
        if not f.endswith('.py'):
            continue
        path = os.path.join(root, f)
        try:
            with open(path, 'r', encoding='utf-8') as fh:
                content = fh.read()
            has_create = 'repo.create(' in content or 'session.add(' in content or 'session.execute(update' in content
            has_commit = 'session.commit()' in content
            if has_create and not has_commit and 'api/' in path:
                print(f'  [WARN] {path}: has DB writes but no session.commit()')
        except:
            pass

# === 6. Hardcoded secrets or IPs ===
print()
print('=' * 60)
print('6. HARDCODED SECRETS / IPs')
print('=' * 60)

for root, dirs, files in os.walk('server'):
    for f in files:
        if not f.endswith('.py'):
            continue
        path = os.path.join(root, f)
        try:
            with open(path, 'r', encoding='utf-8') as fh:
                lines = fh.readlines()
            for i, line in enumerate(lines, 1):
                if re.search(r'password\s*=\s*["\'][^"\']{3,}["\']', line) and 'test' not in line.lower() and 'hash' not in line.lower() and 'Column' not in line:
                    print(f'  [WARN] {path}:{i}: possible hardcoded password')
                if re.search(r'\b10\.0\.\d+\.\d+\b|\b192\.168\.\d+\.\d+\b|\b172\.(1[6-9]|2\d|3[01])\.\d+\.\d+\b', line) and 'example' not in line.lower() and 'comment' not in line.lower():
                    print(f'  [INFO] {path}:{i}: hardcoded private IP')
        except:
            pass

print()
print('=' * 60)
print('SCAN COMPLETE')
print('=' * 60)