251 lines
8.7 KiB
Python
251 lines
8.7 KiB
Python
from fastapi import Depends, HTTPException, status, Request, APIRouter, Path, Query
|
||
from fastapi.security import APIKeyHeader
|
||
from pydantic import BaseModel, Field
|
||
from typing import List, Optional, Dict, Any, Tuple
|
||
from services.device_auth_manager import device_auth_manager
|
||
from utils.logger import session_logger
|
||
from config import settings
|
||
import re
|
||
|
||
router = APIRouter(prefix="/api/auth", tags=["Auth"])
|
||
|
||
device_id_header = APIKeyHeader(name="X-Device-ID", auto_error=False)
|
||
device_serial_header = APIKeyHeader(name="X-Device-Serial", auto_error=False)
|
||
admin_api_key_header = APIKeyHeader(name="X-Admin-API-Key", auto_error=False)
|
||
client_api_key_header = APIKeyHeader(name="X-Client-Key", auto_error=False)
|
||
|
||
async def verify_device(
|
||
device_id: str = Depends(device_id_header),
|
||
serial_number: str = Depends(device_serial_header)
|
||
):
|
||
"""验证设备ID和序列号"""
|
||
if not device_id or not serial_number:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="缺少设备认证信息",
|
||
headers={"WWW-Authenticate": "DeviceAuth"},
|
||
)
|
||
is_valid = await device_auth_manager.authenticate_device(device_id, serial_number)
|
||
if not is_valid:
|
||
session_logger.warning("system", "auth", f"设备 {device_id} 认证失败")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="设备认证失败",
|
||
headers={"WWW-Authenticate": "DeviceAuth"},
|
||
)
|
||
return device_id
|
||
|
||
async def api_auth(
|
||
device_id: str = Depends(verify_device)
|
||
):
|
||
"""API认证依赖项，只使用设备ID和序列号认证"""
|
||
return device_id
|
||
|
||
async def admin_auth(api_key: str = Depends(admin_api_key_header)) -> bool:
|
||
"""验证管理员API密钥"""
|
||
if not api_key:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="缺少管理员API密钥",
|
||
headers={"WWW-Authenticate": "AdminAuth"},
|
||
)
|
||
return api_key == settings.admin_api_key
|
||
|
||
async def client_auth(client_api_key: str = Depends(client_api_key_header)) -> bool:
|
||
"""验证小程序客户端API密钥"""
|
||
if not client_api_key:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="缺少客户端API密钥",
|
||
headers={"WWW-Authenticate": "ClientAuth"},
|
||
)
|
||
|
||
if client_api_key != settings.client_api_key:
|
||
session_logger.warning("system", "auth", "客户端API密钥验证失败")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="客户端API密钥无效",
|
||
headers={"WWW-Authenticate": "ClientAuth"},
|
||
)
|
||
|
||
return True
|
||
|
||
async def admin_or_api_auth(
|
||
device_id: str = Path(...),
|
||
admin_api_key: str = Depends(admin_api_key_header),
|
||
authenticated_device_id: str = Depends(api_auth)
|
||
) -> Tuple[str, bool]:
|
||
"""允许管理员或设备本身访问，返回(认证设备ID, 是否管理员)"""
|
||
is_admin = admin_api_key == settings.admin_api_key if admin_api_key else False
|
||
|
||
if not is_admin and authenticated_device_id != device_id:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail="没有权限查询其他设备",
|
||
headers={"WWW-Authenticate": "DeviceAuth"},
|
||
)
|
||
|
||
return authenticated_device_id, is_admin
|
||
|
||
class DeviceRegistrationRequest(BaseModel):
|
||
device_id: str = Field(..., min_length=16, max_length=64, description="设备ID，基于ESP32的MAC地址生成")
|
||
serial_number: str = Field(..., min_length=18, max_length=64, description="序列号，包含批次前缀和唯一码")
|
||
batch_id: Optional[str] = Field(None, description="批次ID，使用YYYYMMDD格式")
|
||
is_active: bool = Field(True, description="设备激活状态")
|
||
|
||
class DeviceRegistrationResponse(BaseModel):
|
||
status: str
|
||
device_id: str
|
||
serial_number: str
|
||
batch_id: Optional[str] = None
|
||
is_active: bool
|
||
|
||
@router.post("/register-device", response_model=DeviceRegistrationResponse)
|
||
async def register_device(
|
||
request: DeviceRegistrationRequest,
|
||
admin_authenticated: bool = Depends(admin_auth)
|
||
):
|
||
"""注册新设备到认证白名单"""
|
||
batch_id = request.batch_id
|
||
if not batch_id and len(request.serial_number) >= 8:
|
||
batch_id = request.serial_number[:8]
|
||
|
||
success = await device_auth_manager.register_device(
|
||
request.device_id,
|
||
request.serial_number,
|
||
batch_id,
|
||
request.is_active
|
||
)
|
||
|
||
if not success:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="设备注册失败",
|
||
)
|
||
|
||
return {
|
||
"status": "success",
|
||
"device_id": request.device_id,
|
||
"serial_number": request.serial_number,
|
||
"batch_id": batch_id,
|
||
"is_active": request.is_active
|
||
}
|
||
|
||
class BatchDeviceRegistrationRequest(BaseModel):
|
||
devices: List[DeviceRegistrationRequest]
|
||
|
||
class BatchDeviceRegistrationResponse(BaseModel):
|
||
status: str
|
||
registered_count: int
|
||
failed_count: int
|
||
details: List[Dict[str, Any]]
|
||
|
||
@router.post("/register-devices-batch", response_model=BatchDeviceRegistrationResponse)
|
||
async def register_devices_batch(
|
||
request: BatchDeviceRegistrationRequest,
|
||
admin_authenticated: bool = Depends(admin_auth)
|
||
):
|
||
"""批量注册设备到认证白名单"""
|
||
results = []
|
||
success_count = 0
|
||
failed_count = 0
|
||
|
||
for device in request.devices:
|
||
batch_id = device.batch_id
|
||
if not batch_id and len(device.serial_number) >= 8:
|
||
batch_id = device.serial_number[:8]
|
||
|
||
success = await device_auth_manager.register_device(
|
||
device.device_id,
|
||
device.serial_number,
|
||
batch_id,
|
||
device.is_active
|
||
)
|
||
|
||
if success:
|
||
success_count += 1
|
||
results.append({
|
||
"status": "success",
|
||
"device_id": device.device_id,
|
||
"serial_number": device.serial_number,
|
||
"batch_id": batch_id
|
||
})
|
||
else:
|
||
failed_count += 1
|
||
results.append({
|
||
"status": "failed",
|
||
"device_id": device.device_id,
|
||
"serial_number": device.serial_number
|
||
})
|
||
|
||
return {
|
||
"status": "completed",
|
||
"registered_count": success_count,
|
||
"failed_count": failed_count,
|
||
"details": results
|
||
}
|
||
|
||
@router.get("/verify-device/{device_id}")
|
||
async def verify_device(
|
||
device_id: str = Path(...),
|
||
serial_number: str = None,
|
||
auth_result: Tuple[str, bool] = Depends(admin_or_api_auth)
|
||
):
|
||
"""验证设备是否已正确注册 (管理员或设备自身可使用)"""
|
||
authenticated_device_id, is_admin = auth_result
|
||
|
||
if is_admin and serial_number:
|
||
is_valid = await device_auth_manager.authenticate_device(device_id, serial_number)
|
||
elif not is_admin:
|
||
is_valid = True
|
||
else:
|
||
is_valid = None
|
||
|
||
device_info = await device_auth_manager.get_device_info(device_id)
|
||
|
||
return {
|
||
"device_id": device_id,
|
||
"is_valid": is_valid,
|
||
"device_info": device_info
|
||
}
|
||
|
||
class SerialQueryResponse(BaseModel):
|
||
device_id: str
|
||
serial_number: str
|
||
|
||
@router.get("/query-serial", response_model=SerialQueryResponse)
|
||
async def query_serial_by_mac(
|
||
mac_address: str = Query(..., description="设备MAC地址"),
|
||
client_authenticated: bool = Depends(client_auth)
|
||
):
|
||
"""通过MAC地址查询设备ID和序列号"""
|
||
mac_pattern = re.compile(r'^([0-9A-Fa-f]{2}[:-]){5}([0-9A-Fa-f]{2})$')
|
||
if not mac_pattern.match(mac_address):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="MAC地址格式无效，应为XX:XX:XX:XX:XX:XX或XX-XX-XX-XX-XX-XX格式"
|
||
)
|
||
|
||
normalized_mac = mac_address.replace(":", "").replace("-", "").upper()
|
||
|
||
expected_device_id = f"TalkingQ_{normalized_mac}"
|
||
|
||
device_info = await device_auth_manager.get_device_info(expected_device_id)
|
||
|
||
if not device_info or not device_info.get("is_active", False):
|
||
session_logger.warning("system", "auth", f"未找到MAC地址{mac_address}对应的设备或设备未激活")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail="未找到对应的设备信息或设备未激活"
|
||
)
|
||
|
||
return {
|
||
"device_id": expected_device_id,
|
||
"serial_number": device_info["serial_number"]
|
||
}
|
||
|
||
@router.get("/verify")
|
||
async def verify_auth(device_id: str = Depends(api_auth)):
|
||
"""验证设备认证状态"""
|
||
return {"status": "authenticated", "device_id": device_id}
|