设计方案
安全威胁模型
| 威胁 | 对策 |
|---|---|
| 传输嗅探 | XOR 流密码(不做安全保证,仅混淆) + HTTPS |
| 重放攻击 | timestamp + nonce 一次性令牌 |
| 伪造响应 | HMAC-SHA256 签名验证 |
| 暴力穷举卡密 | 卡密 UUID 格式 + 尝试次数限制 |
注意:XOR 流密码仅用于轻量混淆,生产环境通信层请使用 TLS。
数据模型
from sqlalchemy import Column, String, Integer, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
import datetime, uuid
Base = declarative_base()
class Software(Base):
__tablename__ = "softwares"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String, unique=True, nullable=False)
secret = Column(String, nullable=False) # HMAC 密钥(服务端保存)
class Plan(Base):
__tablename__ = "plans"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
software_id = Column(String, nullable=False)
name = Column(String, nullable=False) # "月卡" "年卡" "永久"
days = Column(Integer, nullable=False) # 0=永久
class License(Base):
__tablename__ = "licenses"
key = Column(String, primary_key=True)
plan_id = Column(String, nullable=False)
machine_id = Column(String, nullable=True) # 绑机器码,首次激活写入
activated = Column(Boolean, default=False)
expire_at = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
banned = Column(Boolean, default=False)
卡密生成
import secrets, string
def gen_license_key(prefix: str = "LIC") -> str:
# 生成格式: LIC-XXXX-XXXX-XXXX-XXXX
chars = string.ascii_uppercase + string.digits
parts = ["".join(secrets.choice(chars) for _ in range(4)) for _ in range(4)]
return f"{prefix}-{'-'.join(parts)}"
XOR 流密码
import hashlib
def xor_cipher(data: bytes, key: str) -> bytes:
# XOR 流密码(混淆用,非加密保证)
key_bytes = hashlib.sha256(key.encode()).digest()
result = bytearray(len(data))
for i, b in enumerate(data):
result[i] = b ^ key_bytes[i % len(key_bytes)]
return bytes(result)
FastAPI 验证接口
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
import hashlib, hmac, time, json, base64
app = FastAPI(title="License Server")
class VerifyRequest(BaseModel):
software_id: str
license_key: str
machine_id: str
timestamp: int
nonce: str
signature: str # HMAC-SHA256(software_id+license_key+machine_id+timestamp+nonce, secret)
used_nonces: set = set() # 生产环境用 Redis TTL
@app.post("/api/verify")
async def verify(req: VerifyRequest, db=Depends(get_db)):
# 1. 防重放
if abs(time.time() - req.timestamp) > 60:
raise HTTPException(400, "请求已过期")
if req.nonce in used_nonces:
raise HTTPException(400, "重放攻击检测")
used_nonces.add(req.nonce)
# 2. 查软件
sw = db.query(Software).filter_by(id=req.software_id).first()
if not sw:
raise HTTPException(404, "软件不存在")
# 3. 验证签名
raw = f"{req.software_id}{req.license_key}{req.machine_id}{req.timestamp}{req.nonce}"
expected = hmac.new(sw.secret.encode(), raw.encode(), hashlib.sha256).hexdigest()
if not hmac.compare_digest(expected, req.signature):
raise HTTPException(401, "签名错误")
# 4. 查卡密
lic = db.query(License).filter_by(key=req.license_key).first()
if not lic or lic.banned:
raise HTTPException(403, "卡密无效或已封禁")
# 5. 机器码绑定
if lic.machine_id and lic.machine_id != req.machine_id:
raise HTTPException(403, "机器码不匹配")
if not lic.machine_id:
lic.machine_id = req.machine_id
lic.activated = True
db.commit()
# 6. 检查过期
import datetime
if lic.expire_at and lic.expire_at < datetime.datetime.utcnow():
raise HTTPException(403, "卡密已过期")
result = {"valid": True, "expire_at": str(lic.expire_at) if lic.expire_at else "永久"}
return result
客户端 SDK
import hmac, hashlib, time, uuid, requests, json
class LicenseClient:
def __init__(self, server: str, software_id: str, secret: str):
self.server = server.rstrip("/")
self.software_id = software_id
self.secret = secret
def _sign(self, license_key: str, machine_id: str, ts: int, nonce: str) -> str:
raw = f"{self.software_id}{license_key}{machine_id}{ts}{nonce}"
return hmac.new(self.secret.encode(), raw.encode(), hashlib.sha256).hexdigest()
def verify(self, license_key: str, machine_id: str) -> dict:
ts = int(time.time())
nonce = uuid.uuid4().hex
sig = self._sign(license_key, machine_id, ts, nonce)
r = requests.post(f"{self.server}/api/verify", json={
"software_id": self.software_id,
"license_key": license_key,
"machine_id": machine_id,
"timestamp": ts,
"nonce": nonce,
"signature": sig,
}, timeout=10)
return r.json()
# 使用
client = LicenseClient("https://license.example.com", "sw_id", "shared_secret")
result = client.verify("LIC-ABCD-EFGH-IJKL-MNOP", "machine_fingerprint_hash")
if result.get("valid"):
print(f"验证通过,到期时间: {result['expire_at']}")
管理界面
FastAPI + Jinja2 提供简单的后台管理页面:批量生成卡密、封禁卡密、查看激活记录。