task-283(后台迁移收尾/发布): admin-vue 后台工程入库 + Flask 后台整体退役
- 新后台 admin-frontend-vue 整工程入库(base=/admin-vue/、History 路由、Nginx 静态托管方案与 verify-dist 校验) - Java AdminConsoleController 改为入口重定向: /admin|/admin.html、/login|/login.html -> /admin-vue/; 删除 classpath:static 旧 admin.html/login.html 单页与 admin.js 副本 - Flask 后台整体退役: 删除 admin_api/auth/main 蓝图、web_source 页面、static 脚本与 admin 相关测试; app.py 收敛为仅注册 version_bp(/api/version、/api/version/latest, 供桌面端更新检查) - AdminApiGuardFilterTest 豁免样例路径 /admin.html -> /admin-vue/
This commit is contained in:
@@ -1 +1 @@
|
||||
# Utils 包：数据库、认证装饰器、模板渲染等
|
||||
# Utils 包：数据库连接（管理后台旧代码 task-283 删除后仅保留 db 工具）
|
||||
|
||||
@@ -1,113 +0,0 @@
|
||||
"""
|
||||
认证装饰器与 session 校验
|
||||
"""
|
||||
from functools import wraps
|
||||
|
||||
from flask import request, redirect, url_for, session, jsonify, g
|
||||
|
||||
from utils.db import get_db
|
||||
|
||||
|
||||
def is_session_user_valid():
|
||||
"""校验 session 中的 user_id 是否仍存在；不存在则清空 session。"""
|
||||
uid = session.get('user_id')
|
||||
if not uid:
|
||||
return False
|
||||
cached_user = getattr(g, '_current_user_row', None)
|
||||
if cached_user and cached_user.get('id') == uid:
|
||||
return True
|
||||
try:
|
||||
conn = get_db()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
|
||||
(uid,)
|
||||
)
|
||||
row = cur.fetchone()
|
||||
conn.close()
|
||||
if not row:
|
||||
session.clear()
|
||||
return False
|
||||
g._current_user_row = row
|
||||
return True
|
||||
except Exception:
|
||||
session.clear()
|
||||
return False
|
||||
|
||||
|
||||
def get_current_admin_role():
|
||||
"""返回当前登录用户的管理角色：super_admin / admin / None。"""
|
||||
uid = session.get('user_id')
|
||||
if not uid:
|
||||
return None, None
|
||||
try:
|
||||
row = getattr(g, '_current_user_row', None)
|
||||
if not row or row.get('id') != uid:
|
||||
conn = get_db()
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
|
||||
(uid,)
|
||||
)
|
||||
row = cur.fetchone()
|
||||
conn.close()
|
||||
if not row:
|
||||
return None, None
|
||||
g._current_user_row = row
|
||||
role = (row.get('role') or '').strip().lower()
|
||||
if not role:
|
||||
role = 'super_admin' if row.get('is_admin') and row.get('created_by_id') is None else (
|
||||
'admin' if row.get('is_admin') else 'normal'
|
||||
)
|
||||
return role, row
|
||||
except Exception:
|
||||
return None, None
|
||||
|
||||
|
||||
def is_current_user_admin():
|
||||
role, _ = get_current_admin_role()
|
||||
return role in ('super_admin', 'admin')
|
||||
|
||||
|
||||
def _is_ajax_request():
|
||||
if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
|
||||
return True
|
||||
if request.path.startswith('/api/'):
|
||||
return True
|
||||
accept = (request.headers.get('Accept') or '').lower()
|
||||
if 'application/json' in accept:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def login_required(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
if not session.get('user_id') or not is_session_user_valid():
|
||||
if _is_ajax_request():
|
||||
return jsonify({'success': False, 'error': '未登录'}), 401
|
||||
return redirect(url_for('auth.login'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
|
||||
|
||||
def admin_required(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
if not session.get('user_id') or not is_session_user_valid():
|
||||
if _is_ajax_request():
|
||||
return jsonify({'success': False, 'error': '未登录'}), 401
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
try:
|
||||
role, _ = get_current_admin_role()
|
||||
if role not in ('super_admin', 'admin'):
|
||||
if _is_ajax_request():
|
||||
return jsonify({'success': False, 'error': '需要管理员权限'}), 403
|
||||
return redirect(url_for('auth.login'))
|
||||
except Exception as exc:
|
||||
if _is_ajax_request():
|
||||
return jsonify({'success': False, 'error': '服务器内部错误，请稍后重试'}), 500
|
||||
return redirect(url_for('auth.login'))
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
@@ -1,22 +0,0 @@
|
||||
"""
|
||||
模板渲染：支持加密 HTML 解密后渲染
|
||||
"""
|
||||
import os
|
||||
from flask import render_template, render_template_string
|
||||
|
||||
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def render_html(template_name: str, **context):
|
||||
"""读取 HTML 模板：若为加密文件则先解密，再渲染。未加密或解密失败时按明文渲染。"""
|
||||
path = os.path.join(BASE_DIR, "web_source", template_name)
|
||||
if not os.path.isfile(path):
|
||||
return render_template(template_name, **context)
|
||||
with open(path, "rb") as f:
|
||||
raw = f.read()
|
||||
try:
|
||||
from html_crypto import decrypt
|
||||
content = decrypt(raw).decode("utf-8")
|
||||
except Exception:
|
||||
content = raw.decode("utf-8", errors="replace")
|
||||
return render_template_string(content, **context)
|
||||
@@ -1,58 +0,0 @@
|
||||
"""
|
||||
SSRF 防护：检测 URL 是否指向内网/本机地址，禁止服务端请求。
|
||||
"""
|
||||
import ipaddress
|
||||
import socket
|
||||
from urllib.parse import urlparse
|
||||
|
||||
_PRIVATE_NETWORKS = [
|
||||
ipaddress.ip_network('0.0.0.0/8'),
|
||||
ipaddress.ip_network('10.0.0.0/8'),
|
||||
ipaddress.ip_network('100.64.0.0/10'),
|
||||
ipaddress.ip_network('127.0.0.0/8'),
|
||||
ipaddress.ip_network('169.254.0.0/16'),
|
||||
ipaddress.ip_network('172.16.0.0/12'),
|
||||
ipaddress.ip_network('192.0.0.0/24'),
|
||||
ipaddress.ip_network('192.168.0.0/16'),
|
||||
ipaddress.ip_network('198.18.0.0/15'),
|
||||
ipaddress.ip_network('224.0.0.0/4'),
|
||||
ipaddress.ip_network('240.0.0.0/4'),
|
||||
ipaddress.ip_network('::1/128'),
|
||||
ipaddress.ip_network('fc00::/7'),
|
||||
ipaddress.ip_network('fe80::/10'),
|
||||
]
|
||||
|
||||
_LOCAL_HOSTNAMES = {
|
||||
'localhost',
|
||||
'localhost.localdomain',
|
||||
'metadata.google.internal',
|
||||
'metadata.azure.internal',
|
||||
'169.254.169.254',
|
||||
}
|
||||
|
||||
|
||||
def is_internal_url(url):
|
||||
"""判断 URL 是否解析到内网/本机/保留地址。解析失败视为不可信返回 True。"""
|
||||
if not url or not isinstance(url, str):
|
||||
return True
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme not in ('http', 'https'):
|
||||
return True
|
||||
host = parsed.hostname
|
||||
if not host:
|
||||
return True
|
||||
host_lower = host.lower().rstrip('.')
|
||||
if host_lower in _LOCAL_HOSTNAMES:
|
||||
return True
|
||||
try:
|
||||
infos = socket.getaddrinfo(host, parsed.port or 80)
|
||||
except socket.gaierror:
|
||||
return True
|
||||
for info in infos:
|
||||
try:
|
||||
ip = ipaddress.ip_address(info[4][0])
|
||||
except ValueError:
|
||||
continue
|
||||
if any(ip in network for network in _PRIVATE_NETWORKS):
|
||||
return True
|
||||
return False
|
||||
Reference in New Issue
Block a user