TermiNexus
d600315829
doc：DocManager 新增 QueryDocuments/QuerySeries 返回查询集，视图 get_queryset 改为委托；invite：新增 InviteManager（QuerySet/GetByInviteID/Create）；sve：新增 SveManager（QuerySet/GetByCandidates/Upsert 支持 CreateDefaults）；community：新增 CommunityManager（板块/分类/帖子/回复/点赞共 20 处，ToggleLike 封装点赞切换与计数同步，聚合查询保留原 values_list+annotate）。 视图层不再直接操作 ORM，类级 queryset 属性保留（DRF 必需）；QueryPosts 收 Filters 字典以保持原边界行为。
207 lines
7.7 KiB
Python
207 lines
7.7 KiB
Python
"""模块：SVE 安全漏洞百科视图
|
||
提供漏洞查询、写入、列表筛选功能。
|
||
|
||
SveID 存储规范（用户定案）：5 字节二进制，hex 形式 0x0YYYYNNNNN
|
||
- YYYY：4 位年份（hex 数字，如 2023 → '2023'）
|
||
- NNNNN：5 位数字编号（hex 数字，如 00001）
|
||
例：SVE-2023-00001 ⇄ 字节 02 02 30 00 01（hex 0202300001）
|
||
"""
|
||
|
||
import re
|
||
import datetime
|
||
from rest_framework import viewsets
|
||
from rest_framework.decorators import action
|
||
from rest_framework.response import Response
|
||
from rest_framework.permissions import AllowAny
|
||
from django.db.models import Q
|
||
|
||
from .manager import SveManager
|
||
from .serializers import SVESerializer, SVEListSerializer, SVEWriteSerializer
|
||
|
||
|
||
def _sve_id_to_bytes(sve_id):
|
||
"""SVE-YYYY-NNNNN（或 0x0YYYYNNNNN / 0YYYYNNNNN）→ 5 字节二进制 ID。
|
||
|
||
无效格式抛 ValueError，由调用方转为 400 响应。
|
||
"""
|
||
s = str(sve_id).strip()
|
||
m = re.fullmatch(r'SVE[-_ ]?(\d{4})[-_ ]?(\d+)', s, re.IGNORECASE)
|
||
if m:
|
||
year, count = m.group(1), m.group(2)
|
||
# 编号按十进制解析后补足 5 位（如 00012 → '00012'），拼成 hex 0+年份+编号
|
||
return bytes.fromhex(f'0{year}{int(count):05d}')
|
||
hex_str = s[2:] if s.lower().startswith('0x') else s
|
||
if len(hex_str) == 10 and re.fullmatch(r'[0-9a-fA-F]{10}', hex_str):
|
||
return bytes.fromhex(hex_str)
|
||
raise ValueError(f'无效的 SveID 格式: {sve_id}，期望 SVE-YYYY-NNNNN')
|
||
|
||
|
||
def _format_sve_id(sve_id):
|
||
"""5 字节 ID → SVE-YYYY-NNNNN；兼容 16 字节补零 / UTF-8 文本等存量形态。"""
|
||
if isinstance(sve_id, bytes):
|
||
hex_str = sve_id.hex()
|
||
# 规范形态：10 位 hex 且首字符为 0（0x0YYYYNNNNN）
|
||
if len(sve_id) == 5 and hex_str.startswith('0'):
|
||
return f'SVE-{hex_str[1:5]}-{hex_str[5:]}'
|
||
# 兼容旧补零存储：去掉尾部 \x00 后重试
|
||
stripped = sve_id.rstrip(b'\x00')
|
||
if stripped != sve_id:
|
||
return _format_sve_id(stripped)
|
||
# 兼容 UTF-8 文本形态（旧字符串存储）
|
||
try:
|
||
text = sve_id.decode('utf-8')
|
||
if text.startswith('SVE-'):
|
||
return text
|
||
except UnicodeDecodeError:
|
||
pass
|
||
return hex_str
|
||
return str(sve_id)
|
||
|
||
|
||
def _detail_data(sve):
|
||
"""详情页统一载荷（SveID 转 SVE-YYYY-NNNNN 可读文本）。"""
|
||
cvss_data = sve.SveCvss or {}
|
||
return {
|
||
'SveID': _format_sve_id(sve.SveID),
|
||
'SveName': sve.SveName,
|
||
'SveDescription': sve.SveDescription,
|
||
'SveVulnStatus': sve.SveVulnStatus,
|
||
'SveProducts': sve.SveProducts,
|
||
'SveCvss': sve.SveCvss,
|
||
'CvssBaseScore': cvss_data.get('score', cvss_data.get('baseScore')),
|
||
'CvssBaseSeverity': cvss_data.get('severity', cvss_data.get('baseSeverity')),
|
||
'SveSolutions': sve.SveSolutions,
|
||
'SveReferences': sve.SveReferences,
|
||
'SveRelatedSves': sve.SveRelatedSves,
|
||
'SveCreateTime': sve.SveCreateTime.isoformat() if sve.SveCreateTime else None,
|
||
'SveLastModified': sve.SveLastModified.isoformat() if sve.SveLastModified else None,
|
||
}
|
||
|
||
|
||
class SVEViewSet(viewsets.ModelViewSet):
|
||
queryset = SveManager.QuerySet()
|
||
serializer_class = SVESerializer
|
||
permission_classes = [AllowAny]
|
||
|
||
@action(detail=False, methods=['get', 'post'])
|
||
def list_sve(self, request):
|
||
if request.method == 'POST':
|
||
ser = SVEListSerializer(data=request.data)
|
||
else:
|
||
ser = SVEListSerializer(data=request.query_params)
|
||
ser.IsValid(raise_exception=True)
|
||
data = ser.ValidatedData
|
||
|
||
qs = SveManager.QuerySet()
|
||
Keyword = data.get('Keyword', '')
|
||
if Keyword:
|
||
qs = qs.filter(
|
||
Q(SveName__icontains=Keyword) |
|
||
Q(SveDescription__icontains=Keyword)
|
||
)
|
||
Status = data.get('Status', '')
|
||
if Status:
|
||
qs = qs.filter(SveVulnStatus=Status)
|
||
|
||
CvssMin = data.get('CvssMin', 0.0)
|
||
CvssMax = data.get('CvssMax', 10.0)
|
||
if CvssMin > 0 or CvssMax < 10:
|
||
all_sves = list(qs)
|
||
filtered = []
|
||
for sve in all_sves:
|
||
cvss_data = sve.SveCvss or {}
|
||
score = cvss_data.get('score', cvss_data.get('baseScore', 0))
|
||
if CvssMin <= score <= CvssMax:
|
||
filtered.append(sve)
|
||
qs = filtered
|
||
|
||
Page = data.get('Page', 1)
|
||
PageSize = data.get('PageSize', 20)
|
||
if isinstance(qs, list):
|
||
total = len(qs)
|
||
start = (Page - 1) * PageSize
|
||
end = start + PageSize
|
||
page_items = qs[start:end]
|
||
else:
|
||
total = qs.count()
|
||
start = (Page - 1) * PageSize
|
||
end = start + PageSize
|
||
page_items = qs[start:end]
|
||
|
||
results = []
|
||
for sve in page_items:
|
||
cvss_data = sve.SveCvss or {}
|
||
results.append({
|
||
'SveID': _format_sve_id(sve.SveID),
|
||
'SveName': sve.SveName,
|
||
'SveDescription': (sve.SveDescription or '')[:200],
|
||
'SveVulnStatus': sve.SveVulnStatus,
|
||
'CvssBaseScore': cvss_data.get('score', cvss_data.get('baseScore')),
|
||
'CvssSeverity': cvss_data.get('severity', cvss_data.get('baseSeverity')),
|
||
'SveCreateTime': sve.SveCreateTime.isoformat() if sve.SveCreateTime else None,
|
||
})
|
||
|
||
return Response({
|
||
'code': 0,
|
||
'data': results,
|
||
'total': total,
|
||
'Page': Page,
|
||
'PageSize': PageSize,
|
||
})
|
||
|
||
@action(detail=False, methods=['get', 'post'])
|
||
def get_information(self, request):
|
||
SveID = request.query_params.get('SveID') or request.data.get('SveID')
|
||
if not SveID:
|
||
return Response({'code': 1, 'msg': '缺少SveID'}, status=400)
|
||
try:
|
||
sve_id_bytes = _sve_id_to_bytes(SveID)
|
||
except ValueError as e:
|
||
return Response({'code': 1, 'msg': str(e)}, status=400)
|
||
|
||
# 依次尝试：规范 5 字节 → 旧补零 16 字节 → 旧 UTF-8 文本字节
|
||
candidates = (
|
||
sve_id_bytes,
|
||
SveID.encode('utf-8')[:16].ljust(16, b'\x00'),
|
||
SveID.encode('utf-8'),
|
||
)
|
||
sve = SveManager.GetByCandidates(candidates)
|
||
if sve is None:
|
||
return Response({'code': 1, 'msg': 'SVE不存在'}, status=404)
|
||
|
||
return Response({'code': 0, 'data': _detail_data(sve)})
|
||
|
||
@action(detail=False, methods=['post'])
|
||
def write(self, request):
|
||
ser = SVEWriteSerializer(data=request.data)
|
||
ser.IsValid(raise_exception=True)
|
||
data = ser.ValidatedData
|
||
try:
|
||
sve_id_bytes = _sve_id_to_bytes(data['SveID'])
|
||
except ValueError as e:
|
||
return Response({'code': 1, 'msg': str(e)}, status=400)
|
||
|
||
defaults = {
|
||
'SveName': data.get('SveName', ''),
|
||
'SveDescription': data.get('SveDescription', ''),
|
||
'SveVulnStatus': data.get('SveVulnStatus', ''),
|
||
'SveProducts': data.get('SveProducts', []),
|
||
'SveCvss': data.get('SveCvss', {}),
|
||
'SveSolutions': data.get('SveSolutions', ''),
|
||
'SveReferences': data.get('SveReferences', []),
|
||
'SveRelatedSves': data.get('SveRelatedSves', []),
|
||
'SveLastModified': datetime.datetime.utcnow(),
|
||
}
|
||
|
||
sve, created = SveManager.Upsert(
|
||
sve_id_bytes,
|
||
defaults,
|
||
CreateDefaults={'SveCreateTime': datetime.datetime.utcnow()},
|
||
)
|
||
|
||
return Response({
|
||
'code': 0,
|
||
'msg': '创建成功' if created else '更新成功',
|
||
'data': _detail_data(sve),
|
||
})
|