#!/usr/bin/env python3
"""
设备认证网关 — v2 用 requests 做代理转发
监听 8081，Nginx 将所有动态请求转发至此
架构:
Nginx(80) → 静态资源直接返回
→ 动态请求 → 认证网关(8081) → 检查 Cookie
├─ 已授权 → requests 转发到实际后端
└─ 未授权 → 302 → /device-unauthorized/
"""
import os
import sys
import json
import uuid
import sqlite3
import subprocess
import requests
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs, quote
from datetime import datetime
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import device_auth as auth
AUTH_PORT = 8081
# 后端映射
BACKENDS = {
'obsidian_web': 'http://127.0.0.1:8080',
'hermes_chat': 'http://127.0.0.1:8765',
'hermes_chat_http': 'http://127.0.0.1:8766',
'world_monitor':'http://127.0.0.1:5173',
'hermes_knowledge': 'http://127.0.0.1:8010',
'hermes_webui': 'http://127.0.0.1:8643',
}
# 动态资源 → (后端, 目标路径前缀) 映射（按最长前缀匹配）
DYNAMIC = [
('/doc/', ('obsidian_web', '/doc/')),
('/liangliang/', ('obsidian_web', '/liangliang/')),
('/world/', ('world_monitor', '/world/')),
('/hermes/', ('hermes_knowledge', '/hermes/')),
('/ask', ('hermes_chat', '/chat')), # /ask → /chat
('/chat/config-',('chat_static', '/var/www/chat/')), # 聊天配置JS
('/chat/marked.',('chat_static', '/var/www/chat/')), # marked库
('/chat/index.', ('chat_static', '/var/www/chat/')), # 聊天框架HTML
('/chat/chat-track.',('chat_static', '/var/www/chat/')),# 跟踪脚本
('/_chat/', ('hermes_chat_http', '/chat/')), # /_chat/history → 8766 /chat/history
('/chat/tasks', ('hermes_chat_http', '')), # /chat/tasks → 8766/chat/tasks
('/chat/tree', ('hermes_chat_http', '')), # /chat/tree → 8766/chat/tree
('/chat/history',('hermes_chat_http', '')), # /chat/history → 8766/chat/history
('/chat/prd', ('hermes_chat_http', '')), # /chat/prd → 8766/chat/prd
('/webchat/', ('webchat_static', '/var/www/webchat/')),
('/child-mian/', ('mian_static', '/var/www/child-mian/')),
('/child-liangliang/', ('liangliang_static', '/var/www/child-liangliang/')),
('/health', ('obsidian_web', '/health')),
]
# 直接放行的路径（静态资源）
PASSTHROUGH = [
'/assets/',
'/.well-known/',
]
# 不需要设备检查的内部路径（但走 proxy）
NO_AUTH = [
'/health',
'/_device_auth/',
'/_chat/',
'/chat/history',
'/chat/prd',
'/chat/tasks',
'/chat/tree',
]
requests.packages.urllib3.disable_warnings()
def match_backend(path):
"""匹配路径到后端，返回 (backend_name, target_prefix, matched_prefix) 或 (None, None, None)"""
for prefix, (backend, target_prefix) in sorted(DYNAMIC, key=lambda x: -len(x[0])):
if path == prefix.rstrip('/') or path.startswith(prefix):
return backend, target_prefix, prefix
return None, None, None
class AuthGateway(BaseHTTPRequestHandler):
def do_GET(self):
self._handle()
def do_POST(self):
self._handle()
def do_HEAD(self):
self._handle()
def do_OPTIONS(self):
self._handle()
def do_DELETE(self):
self._handle()
def do_PUT(self):
self._handle()
def _handle(self):
path = urlparse(self.path).path
# ===== 0. 黑名单检查 =====
device_id = self.headers.get('X-Device-ID', '') or self._get_cookie('device_id')
if device_id and auth.is_blacklisted(device_id):
self.send_response(403)
self.send_header('Content-Type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(b'Device blocked')
return
# ===== 1. 设备认证内部 API（不经过代理） =====
if path == '/_device_auth/register':
self._handle_device_register()
return
if path == '/_device_auth/check':
self._handle_device_check()
return
if path == '/_device_auth/fingerprint':
self._handle_fingerprint_check()
return
if path == '/_device_auth/list':
self._handle_device_list()
return
# ===== 1. 放行路径 =====
for p in PASSTHROUGH:
if path.startswith(p):
self._proxy()
return
# ===== 2. 不需要认证的内部路径 =====
for p in NO_AUTH:
if path.startswith(p):
self._proxy()
return
# ===== 3. 设备管理页面（放行，不拦截） =====
if path == '/device-auth/' or path == '/device-auth':
self._serve_device_manager()
return
# ===== 4. 未授权页面（旧兼容，放行） =====
if path == '/device-unauthorized/' or path == '/device-unauthorized':
self._serve_unauthorized_page()
return
# ===== 5. 匹配动态资源 =====
backend, target_prefix, matched_prefix = match_backend(path)
if backend is None:
# 不在动态列表 → 放行（兼容老路径）
self._proxy()
return
# ===== 6. 设备认证 =====
device_id = self.headers.get('X-Device-ID', '') or self._get_cookie('device_id')
if device_id:
resource_type = auth.resolve_type(matched_prefix) or 'unknown'
if auth.is_device_allowed(device_id, resource_type):
# ✅ 授权通过
self._proxy(device_id)
return
# ❌ 未授权 → 返回 403 + 设备指纹采集页面
if not device_id:
device_id = str(uuid.uuid4())
self._set_cookie = device_id
self._record_new_device(device_id, path)
# 返回 403 + 采集页
self._serve_fingerprint_collector(device_id)
def _handle_device_register(self):
"""浏览器注册 device_id + 前端指纹 + 设备画像"""
length = int(self.headers.get('Content-Length', 0) or 0)
body = self.rfile.read(length) if length else b'{}'
try:
data = json.loads(body)
did = data.get('device_id', '')
redirect = data.get('redirect', '/')
fingerprint = data.get('fingerprint', '')
profile = data.get('profile', {})
# 如果有指纹，用指纹查询 / 创建设备身份
if fingerprint:
fp_device_id, is_new = auth.get_or_create_fingerprint(
fingerprint, profile, device_id=did or None
)
did = fp_device_id
if did:
# 生成人类可读的设备描述
desc = auth.generate_device_description(profile)
# 如果是新设备（指纹没见过），加入 pending 并推微信
is_new_device = is_new if fingerprint else True
if is_new_device and not auth.device_exists(did):
ip = self.client_address[0]
ua = self.headers.get('User-Agent', '')
rtype = auth.resolve_type(data.get('redirect', '/')) or 'unknown'
now = datetime.now().isoformat()
conn = sqlite3.connect(auth.DB_PATH)
# 检查是否已有 pending
existing = conn.execute("SELECT 1 FROM pending_auth WHERE device_id=?", (did,)).fetchone()
if not existing:
conn.execute("""
INSERT INTO pending_auth (device_id, fingerprint, user_agent, ip, requested_path, resource_type, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (did, fingerprint, ua[:200], ip, '/fingerprint-register', rtype, now))
conn.commit()
conn.close()
# 推微信
self._push_wechat_new_device(did, desc, ua, ip)
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Set-Cookie', f'device_id={did}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
# 检查是否已授权
authorized = auth.device_exists(did)
self.wfile.write(json.dumps({
'ok': True,
'redirect': redirect,
'device_id': did,
'authorized': authorized,
'description': desc,
'is_new_device': is_new_device,
}).encode())
return
except Exception as e:
print(f"[auth] register error: {e}")
import traceback
traceback.print_exc()
self.send_response(400)
self.end_headers()
self.wfile.write(b'{"ok":false}')
def _handle_fingerprint_check(self):
"""指纹找回：前端用指纹查是否有已授权的设备"""
length = int(self.headers.get('Content-Length', 0) or 0)
body = self.rfile.read(length) if length else b'{}'
try:
data = json.loads(body)
fp = data.get('fingerprint', '')
if fp:
found = auth.find_device_by_fingerprint(fp)
if found:
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Set-Cookie', f'device_id={found["device_id"]}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
self.wfile.write(json.dumps({
'ok': True,
'device_id': found['device_id'],
'description': found.get('description', ''),
}).encode())
return
except:
pass
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(b'{"ok":false}')
def _handle_device_list(self):
"""返回所有已注册设备的列表（含指纹描述、授权状态）"""
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
devices = auth.get_fingerprint_descriptions()
self.wfile.write(json.dumps(devices, ensure_ascii=False).encode())
def _handle_device_check(self):
"""浏览器轮询授权状态"""
length = int(self.headers.get('Content-Length', 0) or 0)
body = self.rfile.read(length) if length else b'{}'
try:
data = json.loads(body)
did = data.get('device_id', '')
redirect = data.get('redirect', '/')
if did and auth.device_exists(did):
# ⚠️ 修复：浏览器传来的完整 UUID 可能跟 Cookie 里的截断版不同
# 确保 Set-Cookie 用完整的 UUID 覆盖任何旧截断版
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Set-Cookie', f'device_id={did}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
self.wfile.write(json.dumps({'ok': True, 'redirect': redirect}).encode())
return
# 二次尝试：如果完整 UUID 没找到，尝试截断前16位匹配（兼容旧数据）
if did and len(did) > 16:
truncated = did[:16]
if auth.device_exists(truncated):
# 截断版已授权，用完整 UUID 覆盖 Cookie 并更新 whitelist
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Set-Cookie', f'device_id={did}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
self.wfile.write(json.dumps({'ok': True, 'redirect': redirect}).encode())
# 后台同步：把完整 UUID 加入 whitelist
import threading
def _sync():
try:
conn = sqlite3.connect(auth.DB_PATH)
row = conn.execute("SELECT label, allowed_types, authorized_by FROM device_whitelist WHERE device_id=?", (truncated,)).fetchone()
if row and not conn.execute("SELECT 1 FROM device_whitelist WHERE device_id=?", (did,)).fetchone():
conn.execute("INSERT OR IGNORE INTO device_whitelist (device_id, label, allowed_types, created_at, authorized_at, authorized_by) VALUES (?, ?, ?, datetime('now'), datetime('now'), ?)", (did, row[0], row[1], row[2]))
conn.commit()
conn.close()
except:
pass
threading.Thread(target=_sync, daemon=True).start()
return
except:
pass
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(b'{"ok":false}')
def _record_new_device(self, device_id, path):
"""记录新设备 + 推微信"""
ip = self.client_address[0]
ua = self.headers.get('User-Agent', '')
fingerprint = ua[:100]
is_new = auth.record_device(device_id, fingerprint, ua, ip, path)
if is_new:
try:
pending = auth.get_pending_list(limit=1)
if pending:
p = pending[0]
msg = (
f"🔐 新设备\n"
f"ID: {p['device_id'][:8]}…\n"
f"UA: {p['user_agent'][:40]}\n"
f"IP: {p['ip']}\n"
f"路径: {p['path']}\n"
f"分类: {p['resource_type']}\n"
f"时间: {p['created_at']}\n\n"
f"回复「授权 {p['device_id'][:8]}」开通"
)
subprocess.Popen(
['hermes', 'send', '--to', 'weixin', msg],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
env={**os.environ, 'HERMES_NO_TOOL_OUTPUT': '1'}
)
auth.mark_notified(p['id'])
except Exception as e:
print(f"[!] 推微信失败: {e}")
def _push_wechat_new_device(self, device_id, description, ua, ip):
"""推送新设备通知到微信，带设备画像描述"""
try:
msg = (
f"🔐 新设备\n"
f"📌 {description or '未知设备'}\n"
f"ID: {device_id[:8]}…\n"
f"IP: {ip}\n\n"
f"回复「授权 {device_id[:8]}」开通"
)
subprocess.Popen(
['hermes', 'send', '--to', 'weixin', msg],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
env={**os.environ, 'HERMES_NO_TOOL_OUTPUT': '1'}
)
except Exception as e:
print(f"[!] 推微信失败: {e}")
def _serve_static(self, root_dir, matched_prefix):
"""直接返回静态文件，根据匹配的前缀剥离 URL 路径"""
path = urlparse(self.path).path
suffix = path[len(matched_prefix):] if path.startswith(matched_prefix) else ''
file_path = root_dir.rstrip('/') + suffix
if not file_path:
file_path = root_dir.rstrip('/') + '/index.html'
if os.path.isdir(file_path):
file_path = file_path.rstrip('/') + '/index.html'
try:
with open(file_path, 'rb') as f:
content = f.read()
ext = os.path.splitext(file_path)[1].lower()
ctype = {
'.html': 'text/html; charset=utf-8',
'.css': 'text/css; charset=utf-8',
'.js': 'application/javascript; charset=utf-8',
'.png': 'image/png',
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.gif': 'image/gif',
'.svg': 'image/svg+xml',
'.ico': 'image/x-icon',
'.json': 'application/json',
'.txt': 'text/plain; charset=utf-8',
}.get(ext, 'application/octet-stream')
self.send_response(200)
self.send_header('Content-Type', ctype)
self.send_header('Content-Length', str(len(content)))
self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate')
self.send_header('Pragma', 'no-cache')
self.send_header('Expires', '0')
self.end_headers()
self.wfile.write(content)
except FileNotFoundError:
self.send_response(404)
self.end_headers()
self.wfile.write(b'Not Found')
def _serve_device_manager(self):
"""设备管理页面：展示所有已授权/待授权设备 + 当前设备采集"""
html = """<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8">
<meta name="viewport" content="width=device-width,initial-scale=1.0">
<title>设备管理</title>
<style>
*{margin:0;padding:0;box-sizing:border-box;}
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;
background:#0a0a0a;color:#e0e0e0;padding:20px;}
h1{font-size:18px;margin-bottom:16px;color:#facc15;}
h2{font-size:14px;color:#888;margin:16px 0 8px;}
.card{background:#111;border-radius:12px;padding:16px;margin-bottom:12px;border:1px solid #222;}
.device{background:#1a1a1a;border-radius:8px;padding:10px;margin-bottom:8px;font-size:12px;line-height:1.8;}
.device .desc{color:#4ade80;font-weight:bold;font-size:13px;}
.device .meta{color:#888;}
.device .auth{color:#facc15;}
.device .unauth{color:#f87171;}
.btn{padding:10px 24px;border:none;border-radius:8px;font-size:14px;cursor:pointer;color:#fff;
width:100%;margin-top:8px;}
.btn-upload{background:#1976d2;}
.btn:disabled{background:#444;cursor:not-allowed;}
.status{padding:8px 12px;border-radius:8px;font-size:13px;margin-top:8px;}
.status.loading{background:#2a2a1e;color:#facc15;}
.status.ok{background:#1a2a1e;color:#4ade80;}
.status.err{background:#2a1a1e;color:#f87171;}
pre{font-size:10px;color:#666;max-height:150px;overflow:auto;margin-top:8px;background:#0f0f0f;padding:8px;border-radius:6px;}
#deviceList{margin-top:12px;}
</style></head>
<body>
<h1>🔐 设备管理</h1>
<div class="card" id="currentDevice">
<div id="myInfo">正在采集当前设备信息...</div>
<div id="uploadStatus" class="status loading">🔍 采集设备信息中...</div>
<button id="uploadBtn" class="btn btn-upload" disabled onclick="uploadFingerprint()">📤 上传此设备到服务器</button>
</div>
<h2>📋 所有已注册设备</h2>
<div class="card" id="deviceList">加载中...</div>
<script>
// ===== 设备指纹采集 =====
function hashString(str){let h=0;for(let i=0;i<str.length;i++){h=((h<<5)-h)+str.charCodeAt(i);h=h&h;}return (h>>>0).toString(16).padStart(8,'0');}
function getCanvasFP(){
const c=document.createElement('canvas');c.width=256;c.height=60;
const x=c.getContext('2d');
x.textBaseline='alphabetic';x.font='16px Arial';
x.fillStyle='#f60';x.fillRect(10,10,40,40);
x.fillStyle='#069';x.fillText('Hermes Device Auth',5,30);
x.fillStyle='rgba(102,204,0,0.7)';x.fillText('Device Manager',15,48);
x.beginPath();x.arc(220,30,18,0,Math.PI*2);
x.strokeStyle='#f00';x.lineWidth=2;x.stroke();
for(let i=0;i<10;i++){x.fillStyle='rgba('+(i*25)+','+(100+i*15)+','+(200-i*10)+',0.5)';x.fillRect(50+i*18,5,8,8);}
return c.toDataURL();
}
function getWebGL(){
try{
const c=document.createElement('canvas');
const g=c.getContext('webgl')||c.getContext('experimental-webgl');
if(!g)return{R:'无',V:'无'};
const e=g.getExtension('WEBGL_debug_renderer_info');
return{R:e?g.getParameter(e.UNMASKED_RENDERER_WEBGL):'不暴露',V:e?g.getParameter(e.UNMASKED_VENDOR_WEBGL):'不暴露'};
}catch(e){return{R:'error',V:'error'};}
}
async function getAudioFP(){
try{
const a=new(window.AudioContext||window.webkitAudioContext)();
const o=a.createOscillator();const n=a.createAnalyser();const g=a.createGain();
o.type='sawtooth';o.frequency.value=440;
o.connect(n);n.connect(g);g.gain.value=0;
const d=new Uint8Array(n.frequencyBinCount);o.start(0);
await new Promise(r=>setTimeout(r,100));
n.getByteFrequencyData(d);let h=0;
for(let i=0;i<100;i++){h=((h<<5)-h)+d[i];h=h&h;}
o.stop();a.close();return h;
}catch(e){return'no-audio';}
}
async function collectProfile(){
const webgl=getWebGL();const audioFp=await getAudioFP();const canvasFp=getCanvasFP();
const ua=navigator.userAgent;
// 系统识别
let os='';let browser='';let browserVersion='';let deviceBrand='';let deviceModel='';
const ul=ua.toLowerCase();
if(/openharmony|harmonyos/i.test(ua)){const v=ua.match(/(?:OpenHarmony|HarmonyOS)\\s*([\\d.]+)/i);os='鸿蒙 '+(v?v[1]:'');}
else if(/android (\\d+[._]\\d+)/i.test(ua)){os='Android '+RegExp.$1.replace('_','.');}
else if(/iphone.*os (\\d+[._]\\d+)/i.test(ua)){os='iOS '+RegExp.$1.replace('_','.');}
else if(/ipad.*os (\\d+[._]\\d+)/i.test(ua)){os='iPadOS '+RegExp.$1.replace('_','.');}
else if(/mac os x (\\d+[._]\\d+)/i.test(ua)){os='macOS '+RegExp.$1.replace('_','.');}
else if(/windows nt 10/i.test(ua))os='Windows 10';else if(/windows nt 11/i.test(ua))os='Windows 11';
else if(/linux/i.test(ua))os='Linux';else os=ua.match(/\\(([^)]+)\\)/)?.[1]||'未知';
// 品牌
if(/huawei|honor|arkweb/i.test(ul))deviceBrand='华为';
else if(/oppo|oneplus|realme/i.test(ul))deviceBrand=/oppo/i.test(ul)?'OPPO':/oneplus/i.test(ul)?'OnePlus':'Realme';
else if(/vivo|iqoo/i.test(ul))deviceBrand=/vivo/i.test(ul)?'vivo':'iQOO';
else if(/xiaomi|mi\\s|redmi|poco/i.test(ul))deviceBrand='小米';
else if(/samsung|sm-/i.test(ul))deviceBrand='三星';
else if(/iphone|ipad/i.test(ul))deviceBrand='Apple';
else if(/macintosh|mac os/i.test(ul))deviceBrand='Mac';
else if(/windows/i.test(ul))deviceBrand='Windows PC';
else if(/linux/i.test(ul))deviceBrand='Linux';
// 型号
if(deviceBrand==='华为'){
const hm=ua.match(/(?:[A-Z]{3,4}-(?:AL|TL|DL|L|N)\\d{2,3}|BRA-AL00|JAD-AL[0-9]{2}|JAD-LX[0-9]{2}|TET-AN[0-9]{2}|NOP-AN[0-9]{2})/i);
if(hm)deviceModel=hm[0].toUpperCase();
const av=ua.match(/ArkWeb\\/([\\d.]+)/);
if(av&&!deviceModel)deviceModel='华为('+av[0]+')';
if(!deviceModel&&/openharmony/i.test(ua))deviceModel='鸿蒙设备';
}else if(deviceBrand==='三星'){const sm=ua.match(/SM-[A-Z0-9]+/i);if(sm)deviceModel=sm[0].toUpperCase();}
else if(deviceBrand==='小米'){const mi=ua.match(/(?:MI\\s\\d[A-Z]*|MIX\\s\\d|Redmi\\s[A-Z0-9]+)/i);if(mi)deviceModel=mi[0];}
else if(/Apple|Mac/i.test(deviceBrand)){const ip=ua.match(/(iPhone\\d+,\\d+|iPad\\d+,\\d+)/i);if(ip)deviceModel=ip[0];}
// 浏览器
if(/micromessenger/i.test(ua)){browser='微信';const w=ua.match(/MicroMessenger\\/([\\d.]+)/i);if(w)browserVersion=w[1];}
else if(/edg/i.test(ua)){browser='Edge';browserVersion=ua.match(/edg\\/([\\d.]+)/i)?.[1]||'';}
else if(/chrome/i.test(ua)){browser='Chrome';browserVersion=ua.match(/chrome\\/([\\d.]+)/i)?.[1]||'';}
else if(/safari/i.test(ua)){browser='Safari';browserVersion=ua.match(/version\\/([\\d.]+)/i)?.[1]||'';}
else if(/firefox/i.test(ua)){browser='Firefox';browserVersion=ua.match(/firefox\\/([\\d.]+)/i)?.[1]||'';}
// 构造profile
const profile={
userAgent:ua,os,browser,browserVersion,deviceBrand,deviceModel,
platform:navigator.platform||'',language:navigator.language,
sec_ch_ua:navigator.userAgentData?.brands?.map(b=>b.brand+' v'+b.version).join(', ')||'',
sec_ch_ua_platform:navigator.userAgentData?.platform||'',
sec_ch_ua_model:navigator.userAgentData?.model||'',
screenWidth:screen.width,screenHeight:screen.height,screenColorDepth:screen.colorDepth,
devicePixelRatio:window.devicePixelRatio||1,orientation:screen.orientation?.type||'',
hardwareConcurrency:navigator.hardwareConcurrency||'未知',deviceMemory:navigator.deviceMemory||'未知',
maxTouchPoints:navigator.maxTouchPoints||0,touchDevice:'ontouchstart' in window,
webglRenderer:webgl.R,webglVendor:webgl.V,audioFingerprint:audioFp,canvasData:canvasFp,
timezone:Intl.DateTimeFormat().resolvedOptions().timeZone,timezoneOffset:new Date().getTimezoneOffset(),
cookiesEnabled:navigator.cookieEnabled,localTime:new Date().toString(),
};
const hashInput=canvasFp+'|'+webgl.R+'|'+webgl.V+'|'+audioFp+'|'+screen.width+'x'+screen.height+'x'+screen.colorDepth+'|'+window.devicePixelRatio+'|'+navigator.hardwareConcurrency;
const fpHash=hashString(hashInput);
return{profile,fpHash};
}
// ===== 上传 =====
async function uploadFingerprint(){
const btn=document.getElementById('uploadBtn');
const st=document.getElementById('uploadStatus');
btn.disabled=true;btn.textContent='⏳ 上传中...';
st.className='status loading';st.textContent='📤 上传到服务器...';
try{
const fp=window.__fp;const profile=window.__profile;
const did=localStorage.getItem('hdid')||'d'+Date.now()+Math.random().toString(36).slice(2);
localStorage.setItem('hdid',did);
const r=await fetch('/_device_auth/register',{
method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({device_id:did,redirect:location.pathname, fingerprint:fp,profile})
});
const d=await r.json();
if(d.ok){
st.className='status ok';
st.textContent='✅ 已上传！设备: '+(d.description||'未知')+(d.authorized?'\\n✅ 已授权':'\\n⏳ 等待管理员授权...');
if(d.authorized){btn.textContent='✅ 已授权';}else{btn.textContent='⏳ 等待授权';setTimeout(loadDeviceList,3000);}
}else{st.className='status err';st.textContent='❌ 上传失败';btn.disabled=false;btn.textContent='🔄 重试';}
}catch(e){st.className='status err';st.textContent='❌ 错误: '+e.message;btn.disabled=false;btn.textContent='🔄 重试';}
}
// ===== 设备列表 =====
async function loadDeviceList(){
try{
const r=await fetch('/_device_auth/list');
const devices=await r.json();
const el=document.getElementById('deviceList');
if(!devices||devices.length===0){el.innerHTML='<div style="color:#666;font-size:13px;">暂无设备记录</div>';return;}
el.innerHTML=devices.map(d=>{
const a=d.authorized?'✅ 已授权':'⏳ 待授权';
const aCls=d.authorized?'auth':'unauth';
return '<div class="device"><div class="desc">'+escHtml(d.description||'未知设备')+'</div>'+
'<div class="meta">编号: '+escHtml(d.device_id?.substring(0,12)||'')+'…</div>'+
'<div class="meta">指纹: '+escHtml(d.fingerprint||'')+'</div>'+
'<div class="meta">最后在线: '+(d.last_seen_at||'')+'</div>'+
'<div class="'+aCls+'">'+a+'</div></div>';
}).join('');
}catch(e){document.getElementById('deviceList').innerHTML='<div style="color:#f87171;">加载失败: '+e.message+'</div>';}
}
function escHtml(s){if(!s)return'';return String(s).replace(/&/g,'&').replace(/</g,'<').replace(/>/g,'>');}
// ===== 主流程 =====
(async function(){
const st=document.getElementById('uploadStatus');
try{
const{profile,fpHash}=await collectProfile();
window.__fp=fpHash;window.__profile=profile;
document.getElementById('myInfo').innerHTML=
'<div style="font-size:12px;line-height:1.8;">'+
'🔑 指纹: <span style="color:#4ade80;font-weight:bold;">'+fpHash+'</span><br>'+
'📱 设备: '+escHtml(profile.deviceBrand||'')+' '+escHtml(profile.deviceModel||'')+'<br>'+
'💻 系统: '+escHtml(profile.os||'')+'<br>'+
'🌐 浏览器: '+escHtml(profile.browser||'')+' '+escHtml(profile.browserVersion||'')+'<br>'+
'🕐 时区: '+escHtml(profile.timezone||'')+'<br>'+
'📐 屏幕: '+profile.screenWidth+'×'+profile.screenHeight+' (x'+profile.devicePixelRatio+')'+
'</div>';
st.className='status ok';st.textContent='✅ 采集完成，点击上传';
document.getElementById('uploadBtn').disabled=false;
// 加载设备列表
loadDeviceList();
}catch(e){st.className='status err';st.textContent='采集失败: '+e.message;}
})();
</script>
</body></html>"""
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
def _serve_fingerprint_collector(self, device_id):
"""403 页面：设备未授权 + 嵌入指纹采集器（极简版，兼容微信浏览器）"""
html = """<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8">
<meta name="viewport" content="width=device-width,initial-scale=1.0">
<title>设备未授权</title>
<style>
*{margin:0;padding:0;box-sizing:border-box;}
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;
background:#0a0a0a;color:#e0e0e0;padding:20px;}
h1{font-size:18px;margin-bottom:4px;color:#facc15;}
.sub{font-size:12px;color:#666;margin-bottom:12px;}
.card{background:#111;border-radius:12px;padding:14px;margin-bottom:10px;border:1px solid #222;}
.fp{font-size:18px;font-weight:bold;color:#4ade80;text-align:center;padding:6px 0;}
.row{display:flex;justify-content:space-between;padding:3px 0;font-size:12px;border-bottom:1px solid #1a1a1a;}
.row:last-child{border-bottom:none;}
.row .l{color:#888;}
.row .v{color:#ddd;word-break:break-all;}
.btn{display:block;padding:14px;border:none;border-radius:10px;font-size:16px;font-weight:600;
cursor:pointer;color:#fff;width:100%;margin-top:8px;text-align:center;}
.btn-up{background:#1976d2;}
.btn:disabled{background:#444;cursor:not-allowed;}
.st{padding:8px 12px;border-radius:8px;font-size:13px;margin-top:8px;text-align:center;}
.st.loading{background:#2a2a1e;color:#facc15;}
.st.ok{background:#1a2a1e;color:#4ade80;}
.st.err{background:#2a1a1e;color:#f87171;}
</style></head>
<body>
<h1>🔒 此设备未授权</h1>
<p class="sub">点击按钮将设备信息发送给管理员，授权后自动放行</p>
<div class="card"><div class="fp" id="fpD">计算中…</div></div>
<div class="card"><div id="infoD"></div></div>
<div class="card"><div id="hwD"></div></div>
<div class="card"><div id="netD"></div></div>
<div class="card">
<div id="stD" class="st loading">🔍 采集设备信息…</div>
<button id="upBtn" class="btn btn-up" disabled>📤 上传并请求授权</button>
</div>
<script>
(function(){
var did = localStorage.getItem('hdid');
if(!did){did='d'+Date.now()+Math.random().toString(36).slice(2);localStorage.setItem('hdid',did);}
var redir = location.pathname+location.search;
function esc(s){if(!s)return'';return String(s).replace(/&/g,'&').replace(/</g,'<').replace(/>/g,'>');}
function r(l,v){return '<div class=\"row\"><span class=\"l\">'+l+'</span><span class=\"v\">'+v+'</span></div>';}
function hash(s){var h=0;for(var i=0;i<s.length;i++){h=((h<<5)-h)+s.charCodeAt(i);h=h&h;}return (h>>>0).toString(16).padStart(8,'0');}
function getCFP(){
try{
var c=document.createElement('canvas');c.width=200;c.height=40;
var x=c.getContext('2d');
x.textBaseline='top';x.font='12px Arial';
x.fillStyle='#f60';x.fillRect(5,5,20,20);
x.fillStyle='#069';x.fillText('Fp'+Date.now(),5,5);
x.fillStyle='rgba(102,204,0,0.5)';x.fillText('DeviceAuth',10,20);
return c.toDataURL();
}catch(e){return 'no-canvas';}
}
function collectBasic(){
var fp={};
var ua=navigator.userAgent;
fp.ua=ua;
if(/openharmony|harmonyos/i.test(ua)) fp.os='鸿蒙';
else if(/android/i.test(ua)) fp.os='Android';
else if(/iphone|ipad/i.test(ua)) fp.os='iOS/iPadOS';
else if(/mac os/i.test(ua)) fp.os='macOS';
else if(/windows/i.test(ua)) fp.os='Windows';
else if(/linux/i.test(ua)) fp.os='Linux';
if(/huawei|honor|arkweb/i.test(ua)) fp.brand='华为';
else if(/xiaomi|mi\s|redmi/i.test(ua)) fp.brand='小米';
else if(/samsung|sm-/i.test(ua)) fp.brand='三星';
else if(/iphone|ipad/i.test(ua)) fp.brand='Apple';
else if(/macintosh/i.test(ua)) fp.brand='Mac';
else if(/windows/i.test(ua)) fp.brand='Windows PC';
else if(/linux/i.test(ua)) fp.brand='Linux';
if(fp.brand==='华为'){
var hm=ua.match(/(?:[A-Z]{3,4}-(?:AL|TL|DL|L|N)\d{2,3})/i);
if(hm) fp.model=hm[0].toUpperCase();
var av=ua.match(/ArkWeb\/([\d.]+)/);
if(av&&!fp.model) fp.model='华为('+av[0]+')';
}
if(fp.brand==='三星'){var sm=ua.match(/SM-[A-Z0-9]+/i);if(sm)fp.model=sm[0].toUpperCase();}
if(fp.brand==='小米'){var mi=ua.match(/(?:MI\s\d[A-Z]*|MIX|Redmi\s[A-Z0-9]+)/i);if(mi)fp.model=mi[0];}
fp.screen=screen.width+'x'+screen.height;
fp.dpr=window.devicePixelRatio||1;
fp.touch='ontouchstart' in window;
fp.tz=Intl.DateTimeFormat().resolvedOptions().timeZone;
fp.lang=navigator.language;
fp.cores=navigator.hardwareConcurrency||'?';
fp.platform=navigator.platform||'';
fp.browser='';
if(/micromessenger/i.test(ua)) fp.browser='微信';
else if(/edg/i.test(ua)) fp.browser='Edge';
else if(/chrome/i.test(ua)) fp.browser='Chrome';
else if(/safari/i.test(ua)) fp.browser='Safari';
else if(/firefox/i.test(ua)) fp.browser='Firefox';
fp.bwVer=(ua.match(/(?:Chrome|Edg|Firefox|Version)\/([\d.]+)/)||[])[1]||'';
fp.wxVer=(ua.match(/MicroMessenger\/([\d.]+)/)||[])[1]||'';
fp.wechat=fp.wxVer?'是('+fp.wxVer+')':'否';
var cfp=getCFP();
fp.cfp=cfp;
var raw=cfp+'|'+fp.screen+'|'+fp.dpr+'|'+(fp.cores||'');
fp.hash=hash(raw);
return fp;
}
async function doCollect(){
try{
var fp=collectBasic();
document.getElementById('fpD').textContent=fp.hash;
document.getElementById('infoD').innerHTML=
r('📱 品牌',esc(fp.brand||'?'))+r('📱 型号',esc(fp.model||'?'))+
r('💻 系统',esc(fp.os||'?'))+
r('🌐 浏览器',esc(fp.browser+(fp.bwVer?' '+fp.bwVer:'')))+
r('🗴 微信',fp.wechat)+
r('🕘 时区',esc(fp.tz||'?'))+
r('📐 分辨率',fp.screen+' (x'+fp.dpr+')')+
r('🗣 语言',esc(fp.lang||'?'));
document.getElementById('hwD').innerHTML=
r('🧠 CPU',String(fp.cores||'?'))+
r('📱 触控',String(fp.touch));
try{
var r2=await fetch('https://ipapi.co/json/');
if(r2.ok){var d2=await r2.json();
fp.ip=d2.ip;fp.loc=[d2.city,d2.region,d2.country_name].filter(Boolean).join(' ');
fp.isp=d2.org||'';
document.getElementById('netD').innerHTML=r('🌐 IP',esc(fp.ip||''))+r('📍 归属地',esc(fp.loc||''));}
}catch(e){}
window.__fp=fp.hash;
window.__profile=fp;
var st=document.getElementById('stD');
st.className='st ok';
st.textContent='✅ 采集完成，点击上传';
document.getElementById('upBtn').disabled=false;
document.getElementById('upBtn').onclick=uploadFP;
}catch(e){
var st=document.getElementById('stD');
st.className='st err';
st.textContent='采集失败: '+e.message;
document.getElementById('upBtn').disabled=false;
document.getElementById('upBtn').textContent='📤 紧急上传(仅UA)';
document.getElementById('upBtn').onclick=emergencyUpload;
}
}
async function uploadFP(){
var btn=document.getElementById('upBtn');var st=document.getElementById('stD');
btn.disabled=true;btn.textContent='⏳ 上传中...';
st.className='st loading';st.textContent='📤 正在上传...';
try{
var r=await fetch('/_device_auth/register',{
method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({device_id:did,redirect:redir,fingerprint:window.__fp,profile:window.__profile})
});
var d=await r.json();
if(d.ok){
st.className='st ok';
st.textContent='✅ 已发送，等待授权...\n设备: '+(d.description||'');
btn.textContent='⏳ 等待授权';
pollAuth();
}else{st.className='st err';st.textContent='❌ 上传失败';btn.disabled=false;btn.textContent='🔄 重试';}
}catch(e){st.className='st err';st.textContent='❌ '+e.message;btn.disabled=false;btn.textContent='🔄 重试';}
}
async function emergencyUpload(){
window.__profile={ua:navigator.userAgent,emergency:true};
window.__fp='emergency-'+Date.now();
await uploadFP();
}
async function pollAuth(){
try{
var r=await fetch('/_device_auth/check',{
method:'POST',headers:{'Content-Type':'application/json'},
body:JSON.stringify({device_id:did,redirect:redir})
});
var d=await r.json();
if(d.ok){
document.getElementById('stD').className='st ok';
document.getElementById('stD').textContent='✅ 已授权！即将跳转...';
document.cookie='device_id='+did+';path=/;max-age='+(365*86400);
setTimeout(function(){location.href=d.redirect||'/';},1200);
return;
}
}catch(e){}
setTimeout(pollAuth,3000);
}
doCollect();
})();
</script>
</body></html>"""
self.send_response(403)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.send_header('Set-Cookie', f'device_id={device_id}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
def _serve_unauthorized_page(self):
"""设备未授权页面"""
qs = parse_qs(urlparse(self.path).query)
redirect = qs.get('redirect', ['/'])[0]
device_id = qs.get('device_id', [''])[0]
html = f"""<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8">
<meta name="viewport" content="width=device-width,initial-scale=1.0">
<title>设备未授权</title>
<style>
*{{margin:0;padding:0;box-sizing:border-box;}}
body{{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;
background:#0a0a0a;color:#e0e0e0;height:100vh;display:flex;
align-items:center;justify-content:center;}}
.card{{background:#111;border-radius:16px;padding:40px;max-width:420px;
text-align:center;border:1px solid #222;}}
.icon{{font-size:64px;margin-bottom:20px;}}
h1{{font-size:20px;margin-bottom:12px;color:#facc15;}}
p{{font-size:14px;color:#888;line-height:1.8;}}
.did{{font-size:11px;color:#555;margin-top:20px;word-break:break-all;
padding:8px;background:#1a1a1a;border-radius:8px;}}
.s{{display:inline-block;margin-top:20px;padding:8px 20px;
border-radius:20px;font-size:13px;background:#2a2a1e;color:#facc15;}}
.s.ok{{background:#1a2a1e;color:#4ade80;}}
</style></head>
<body><div class="card">
<div class="icon">🔐</div>
<h1>本设备未授权</h1>
<p>已通知管理员<br>授权通过后自动跳转</p>
<div class="did" id="did">{device_id[:16] or '生成中...'}</div>
<div class="s" id="st">⏳ 等待授权...</div>
</div>
<script>
let did = localStorage.getItem('hdid');
if(!did){{did=(crypto.randomUUID||(()->'d'+Date.now()+Math.random().toString(36).slice(2)))();localStorage.setItem('hdid',did);}}
document.getElementById('did').textContent=did;
// 注册 device
fetch('/_device_auth/register',{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify({{device_id:did,redirect:'{redirect}'}})}}).catch(()=>{{}});
// 轮询
async function poll(){{
try{{
let r=await fetch('/_device_auth/check',{{method:'POST',headers:{{'Content-Type':'application/json'}},body:JSON.stringify({{device_id:did}})}});
let d=await r.json();
if(d.ok){{
document.getElementById('st').textContent='✅ 已授权，跳转中...';
document.getElementById('st').className='s ok';
document.cookie='device_id='+did+';path=/;max-age='+(365*86400);
setTimeout(()=>location.href=d.redirect||'/',1200);
}}
}}catch(e){{}}
setTimeout(poll,3000);
}}
setTimeout(poll,2000);
</script></body></html>"""
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
if device_id:
self.send_header('Set-Cookie', f'device_id={device_id}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
self.wfile.write(html.encode('utf-8'))
def _proxy(self, device_id=None):
"""代理请求到后端"""
# 找后端
path = urlparse(self.path).path
backend, target_prefix, matched_prefix = match_backend(path)
# ═══ 调试日志 ═══
print(f"[PROXY] path={path} matched=({matched_prefix})→{backend} target_prefix={target_prefix}", flush=True)
# ═══ ═══
if not backend:
# 没匹配到 → 默认走 Flask
backend = 'obsidian_web'
target_prefix = path
matched_prefix = path # fallback: use path directly
# 特殊处理：静态文件直接返回
if backend == 'webchat_static':
self._serve_static(target_prefix, matched_prefix)
return
if backend == 'mian_static':
self._serve_static(target_prefix, matched_prefix)
return
if backend == 'liangliang_static':
self._serve_static(target_prefix, matched_prefix)
return
if backend == 'chat_static':
self._serve_static(target_prefix, matched_prefix)
return
target = BACKENDS.get(backend)
if not target:
self.send_response(502)
self.end_headers()
self.wfile.write(b'Bad Gateway: unknown backend')
return
# 构造目标 URL（替换路径前缀以实现 /ask → /chat 等映射）
qs = urlparse(self.path).query
# matched_prefix and target_prefix are from match_backend(); in fallback both equal path
suffix = ''
if path.startswith(matched_prefix):
suffix = path[len(matched_prefix):]
url = target + target_prefix + suffix
# target_prefix 为空时保持原始路径
if not target_prefix:
url = target.rstrip('/') + '/' + path.lstrip('/')
url = url.rstrip('/')
if qs:
url += '?' + qs
# 读取 body
body = None
if self.command in ('POST', 'PUT', 'PATCH', 'DELETE'):
length = int(self.headers.get('Content-Length', 0) or 0)
if length > 0:
body = self.rfile.read(length)
# 构造 headers（去掉 hop-by-hop）
headers = {}
skip = {'host', 'content-length', 'transfer-encoding', 'connection', 'x-device-id'}
for k, v in self.headers.items():
if k.lower() not in skip:
headers[k] = v
# 透传原始 IP
headers['X-Real-IP'] = self.client_address[0]
headers['X-Forwarded-For'] = self.headers.get('X-Forwarded-For', '') + ',' + self.client_address[0]
try:
resp = requests.request(
method=self.command,
url=url,
data=body,
headers=headers,
timeout=120,
stream=True,
allow_redirects=False,
)
self.send_response(resp.status_code)
# 回传响应头
for k, v in resp.headers.items():
if k.lower() not in ('transfer-encoding', 'content-encoding'):
self.send_header(k, v)
# 设置 Cookie（如果有新 device_id）
if device_id:
self.send_header('Set-Cookie', f'device_id={device_id}; Path=/; Max-Age=31536000; SameSite=Lax')
self.end_headers()
# 流式回传
for chunk in resp.iter_content(chunk_size=65536):
if chunk:
self.wfile.write(chunk)
self.wfile.flush()
except requests.exceptions.Timeout:
self.send_response(504)
self.end_headers()
self.wfile.write(b'Gateway Timeout')
except requests.exceptions.ConnectionError as e:
self.send_response(502)
self.end_headers()
self.wfile.write(f'Backend connection failed: {e}'.encode())
except Exception as e:
self.send_response(502)
self.end_headers()
self.wfile.write(f'Proxy error: {e}'.encode())
def _get_cookie(self, name):
cookies = self.headers.get('Cookie', '')
for c in cookies.split(';'):
c = c.strip()
if c.startswith(name + '='):
return c[len(name)+1:]
return ''
def log_message(self, fmt, *args):
path = urlparse(self.path).path
did = self._get_cookie('device_id') or self.headers.get('X-Device-ID', '')[:8]
print(f"[AUTH] {self.command} {path} -> {args[1]} {args[2]} did={did}")
def main():
print(f"\n{'='*50}")
print(f" 设备认证网关 (port {AUTH_PORT})")
print(f" 动态请求 → 检查 Cookie → 授权放行 / 未授权拦截")
print(f" 白名单: device_auth.db")
print(f"{'='*50}\n")
server = HTTPServer(('0.0.0.0', AUTH_PORT), AuthGateway)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n网关已关闭")
server.server_close()
if __name__ == '__main__':
main()