#!/usr/bin/env python3
"""
设备白名单数据库模块
- 设备标识用浏览器生成的 UUID + 简单指纹
- 白名单只通过微信/阿里云命令行授权
- 无任何登录界面
"""
import sqlite3
import os
import json
from datetime import datetime
DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'device_auth.db')
# ============ 资源分类 ============
# 每个类别路径前缀 + 说明
RESOURCE_TYPES = {
'public': {'paths': ['/notes/'], 'desc': '公开知识库'},
'doc': {'paths': ['/doc/'], 'desc': '个人文档'},
'private': {'paths': ['/private/'], 'desc': '私有知识库'},
'webchat': {'paths': ['/webchat/'], 'desc': '网页聊天'},
'mian': {'paths': ['/child-mian/'], 'desc': '阿棉的聊天'},
'liangliang':{'paths': ['/child-liangliang/'], 'desc': '亮亮的聊天'},
'world': {'paths': ['/world/'], 'desc': '世界监控'},
'admin': {'paths': ['/chat/', '/monitor/', '/hermes/'], 'desc': '管理后台'},
'api': {'paths': ['/api/', '/ask'], 'desc': 'API 接口'},
}
# 反向映射：路径前缀 → 资源类型
PATH_TO_TYPE = {}
for tname, tinfo in RESOURCE_TYPES.items():
for p in tinfo['paths']:
PATH_TO_TYPE[p] = tname
def resolve_type(path):
"""根据请求路径返回资源类型名称"""
for prefix, tname in sorted(PATH_TO_TYPE.items(), key=lambda x: -len(x[0])):
if path.startswith(prefix):
return tname
return None
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS device_whitelist (
device_id TEXT PRIMARY KEY,
fingerprint TEXT DEFAULT '',
label TEXT DEFAULT '',
allowed_types TEXT DEFAULT '*', -- 逗号分隔，* 表示全部
created_at TEXT NOT NULL,
authorized_at TEXT,
authorized_by TEXT DEFAULT '', -- 'wechat' 或 'terminal'
last_seen_at TEXT,
last_seen_path TEXT DEFAULT ''
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS pending_auth (
id INTEGER PRIMARY KEY AUTOINCREMENT,
device_id TEXT NOT NULL,
fingerprint TEXT DEFAULT '',
user_agent TEXT DEFAULT '',
ip TEXT DEFAULT '',
requested_path TEXT DEFAULT '',
resource_type TEXT DEFAULT '',
notified INTEGER DEFAULT 0, -- 是否已推送到微信
created_at TEXT NOT NULL
)
""")
conn.commit()
conn.close()
def device_exists(device_id):
conn = sqlite3.connect(DB_PATH)
row = conn.execute("SELECT 1 FROM device_whitelist WHERE device_id=?", (device_id,)).fetchone()
conn.close()
return row is not None
def is_device_allowed(device_id, resource_type):
"""检查设备是否有权访问某类资源"""
conn = sqlite3.connect(DB_PATH)
row = conn.execute("SELECT allowed_types FROM device_whitelist WHERE device_id=?", (device_id,)).fetchone()
conn.close()
if not row:
return False
allowed = row[0]
if allowed == '*':
return True
types = [t.strip() for t in allowed.split(',')]
return resource_type in types
def record_device(device_id, fingerprint, user_agent, ip, path):
"""记录新设备访问，返回是否是新设备（之前未见过的）"""
conn = sqlite3.connect(DB_PATH)
# 更新 last_seen
now = datetime.now().isoformat()
conn.execute("""
UPDATE device_whitelist SET
last_seen_at=?, last_seen_path=?,
fingerprint=CASE WHEN fingerprint='' THEN ? ELSE fingerprint END
WHERE device_id=?
""", (now, path, fingerprint, device_id))
row = conn.execute("SELECT 1 FROM device_whitelist WHERE device_id=?", (device_id,)).fetchone()
if row:
# 已有设备，检查此次请求的资源类型是否被允许
conn.close()
return False # 不是新设备
# 新设备：加入 pending
rtype = resolve_type(path) or 'unknown'
conn.execute("""
INSERT INTO pending_auth (device_id, fingerprint, user_agent, ip, requested_path, resource_type, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (device_id, fingerprint, user_agent[:200], ip, path, rtype, now))
conn.commit()
conn.close()
return True # 是新设备
def get_pending_list(limit=10):
"""获取待授权的设备列表"""
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("""
SELECT id, device_id, fingerprint, user_agent, ip, requested_path, resource_type, created_at
FROM pending_auth WHERE notified=0 ORDER BY id DESC LIMIT ?
""", (limit,)).fetchall()
conn.close()
return [
{
'id': r[0], 'device_id': r[1], 'fingerprint': r[2],
'user_agent': r[3], 'ip': r[4], 'path': r[5],
'resource_type': r[6], 'created_at': r[7]
}
for r in rows
]
def authorize_device(device_id, label='', allowed_types='*', authorized_by='wechat'):
"""授权设备，同时从 pending 移除"""
conn = sqlite3.connect(DB_PATH)
now = datetime.now().isoformat()
conn.execute("""
INSERT OR REPLACE INTO device_whitelist
(device_id, label, allowed_types, created_at, authorized_at, authorized_by)
VALUES (?, ?, ?, COALESCE((SELECT created_at FROM device_whitelist WHERE device_id=?), ?), ?, ?)
""", (device_id, label, allowed_types, device_id, now, now, authorized_by))
conn.execute("DELETE FROM pending_auth WHERE device_id=?", (device_id,))
conn.commit()
conn.close()
def revoke_device(device_id):
conn = sqlite3.connect(DB_PATH)
conn.execute("DELETE FROM device_whitelist WHERE device_id=?", (device_id,))
conn.execute("DELETE FROM pending_auth WHERE device_id=?", (device_id,))
conn.commit()
conn.close()
def list_devices():
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("""
SELECT device_id, label, allowed_types, authorized_at, authorized_by, last_seen_at, last_seen_path
FROM device_whitelist ORDER BY authorized_at DESC
""").fetchall()
conn.close()
return [
{'device_id': r[0], 'label': r[1], 'allowed_types': r[2],
'authorized_at': r[3], 'authorized_by': r[4],
'last_seen_at': r[5], 'last_seen_path': r[6]}
for r in rows
]
def mark_notified(pending_id):
conn = sqlite3.connect(DB_PATH)
conn.execute("UPDATE pending_auth SET notified=1 WHERE id=?", (pending_id,))
conn.commit()
conn.close()
# ============ 设备画像 ============
# fingerprint_profile 存储前端采集的详细设备信息，指纹哈希作为主键
# profile 字段是 JSON，包含操作系统、浏览器、GPU、屏幕、时区、语言等
# description 字段是人类可读的设备描述（如 "华为 Mate 60 Pro · 上海"）
def get_or_create_fingerprint(fingerprint, profile, device_id=None):
"""根据指纹查找设备；如果指纹不存在则创建记录，返回 device_id"""
conn = sqlite3.connect(DB_PATH)
# 查找已有的指纹
row = conn.execute(
"SELECT device_id, profile FROM device_fingerprints WHERE fingerprint=?",
(fingerprint,)
).fetchone()
if row:
# 已有此指纹 → 更新 profile（如果 device_id 变了就覆盖）
existing_did = row[0]
if device_id and device_id != existing_did:
# 同一个指纹出现了新的 device_id → 用现有的 device_id
pass # 不改变，保持首次绑定的 device_id
# 更新 profile 和 last_seen
conn.execute(
"UPDATE device_fingerprints SET profile=?, last_seen_at=datetime('now') WHERE fingerprint=?",
(json.dumps(profile, ensure_ascii=False), fingerprint)
)
conn.commit()
conn.close()
return existing_did, False # False = 不是新设备
# 新指纹
if not device_id:
import uuid
device_id = str(uuid.uuid4())
# 生成人类可读描述
desc = generate_device_description(profile)
conn.execute("""
INSERT INTO device_fingerprints (fingerprint, device_id, profile, description, first_seen_at, last_seen_at)
VALUES (?, ?, ?, ?, datetime('now'), datetime('now'))
""", (fingerprint, device_id, json.dumps(profile, ensure_ascii=False), desc))
conn.commit()
conn.close()
return device_id, True # True = 新设备
def find_device_by_fingerprint(fingerprint):
"""通过指纹查找已授权的 device_id"""
conn = sqlite3.connect(DB_PATH)
row = conn.execute("""
SELECT w.device_id, f.description
FROM device_fingerprints f
JOIN device_whitelist w ON f.device_id = w.device_id
WHERE f.fingerprint=?
""", (fingerprint,)).fetchone()
conn.close()
if row:
return {'device_id': row[0], 'description': row[1]}
return None
def find_fingerprint_by_device_id(device_id):
"""通过 device_id 查找对应的指纹和描述"""
conn = sqlite3.connect(DB_PATH)
row = conn.execute(
"SELECT fingerprint, profile, description FROM device_fingerprints WHERE device_id=? ORDER BY last_seen_at DESC LIMIT 1",
(device_id,)
).fetchone()
conn.close()
if row:
try:
profile = json.loads(row[1]) if row[1] else {}
except:
profile = {}
return {'fingerprint': row[0], 'profile': profile, 'description': row[2]}
return None
def generate_device_description(profile):
"""根据设备画像生成人类可读的设备描述"""
parts = []
# 设备型号
device_model = ''
if profile.get('sec_ch_ua_model'):
device_model = profile['sec_ch_ua_model']
elif profile.get('deviceBrand') and profile.get('deviceModel'):
device_model = f"{profile['deviceBrand']} {profile['deviceModel']}"
if device_model:
parts.append(device_model)
# 操作系统
os_info = profile.get('os', '')
if not os_info and profile.get('sec_ch_ua_platform'):
os_info = profile['sec_ch_ua_platform']
if os_info:
parts.append(os_info)
# 地理位置（IP 归属地）
if profile.get('ip_location'):
parts.append(profile['ip_location'])
# 设备特征
features = []
if profile.get('touchDevice'):
features.append('📱触控屏')
if profile.get('maxTouchPoints', 0) > 0:
pass # 触控信息通过 emoji 表达
if profile.get('cookiesEnabled') == False:
features.append('🚫Cookie禁用')
if features:
parts.extend(features)
if parts:
return ' · '.join(parts)
return '未知设备'
def get_fingerprint_descriptions():
"""获取所有指纹记录的设备画像（用于微信推送时展示）"""
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("""
SELECT f.fingerprint, f.device_id, f.description, f.profile, f.last_seen_at,
CASE WHEN w.device_id IS NOT NULL THEN 1 ELSE 0 END as authorized
FROM device_fingerprints f
LEFT JOIN device_whitelist w ON f.device_id = w.device_id
ORDER BY f.last_seen_at DESC
""").fetchall()
conn.close()
result = []
for r in rows:
try:
profile = json.loads(r[3]) if r[3] else {}
except:
profile = {}
result.append({
'fingerprint': r[0][:16] + '…' if len(r[0]) > 16 else r[0],
'device_id': r[1],
'description': r[2],
'authorized': bool(r[5]),
'last_seen_at': r[4],
'detail': profile,
})
return result
# ============ 黑名单 ============
def is_blacklisted(device_id):
"""检查设备是否在黑名单中"""
conn = sqlite3.connect(DB_PATH)
row = conn.execute("SELECT 1 FROM blacklist WHERE device_id=?", (device_id,)).fetchone()
conn.close()
return row is not None
def blacklist_device(device_id, reason=''):
conn = sqlite3.connect(DB_PATH)
conn.execute("INSERT OR IGNORE INTO blacklist (device_id, reason) VALUES (?, ?)", (device_id, reason))
conn.commit()
conn.close()
def unblacklist_device(device_id):
conn = sqlite3.connect(DB_PATH)
conn.execute("DELETE FROM blacklist WHERE device_id=?", (device_id,))
conn.commit()
conn.close()
def list_blacklist():
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("SELECT device_id, reason, created_at FROM blacklist ORDER BY created_at DESC").fetchall()
conn.close()
return rows
# 初始化（确保 blacklist 表存在）
_init_conn = sqlite3.connect(DB_PATH)
_init_conn.execute("""
CREATE TABLE IF NOT EXISTS blacklist (
device_id TEXT PRIMARY KEY,
reason TEXT DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
_init_conn.execute("""
CREATE TABLE IF NOT EXISTS device_fingerprints (
fingerprint TEXT PRIMARY KEY,
device_id TEXT NOT NULL,
profile TEXT DEFAULT '{}',
description TEXT DEFAULT '',
first_seen_at TEXT NOT NULL,
last_seen_at TEXT NOT NULL
)
""")
_init_conn.execute("""
CREATE INDEX IF NOT EXISTS idx_fingerprints_device_id ON device_fingerprints(device_id)
""")
_init_conn.commit()
_init_conn.close()
# 初始化
init_db()
if __name__ == '__main__':
# 测试
print("设备白名单数据库已初始化")
print(f"数据库路径: {DB_PATH}")
print(f"资源分类: {', '.join(RESOURCE_TYPES.keys())}")