feat: 添加后台登录验证码、记住密码功能，优化媒体资源与前端规范
- 新增后台登录图形验证码功能，完善登录安全防护 - 新增登录rememberMe参数，控制Refresh Token的会话持久化策略 - 实现OSS私有桶媒体URL自动签名，统一处理图片资源的临时访问签名 - 新增素材库数据库表与上传API，规范媒体资源管理流程 - 统一前端UI图标使用@element-plus/icons-vue，重构布局图标组件 - 登录页新增验证码输入、刷新功能，添加账号记忆与记住密码逻辑 - 更新全套文档，补充API契约、技术决策记录与集成流程说明 - 修复多个业务页面的图标展示问题，新增认证流程相关测试用例
This commit is contained in:
duanshuwen committed
1 parent
2c8c327de7
commit
6245159e7c
35 files changed
+914
-79
No files matched your search
@@ -76,6 +76,32 @@ def test_in_memory_store_supports_login_limit_and_permission_cache():
|
||||
assert store.get_permission_context("admin-1") == {"permissions": ["admin:read"]}
|
||||
|
||||
|
||||
def test_in_memory_captcha_is_single_use():
|
||||
store = InMemoryAdminSessionStore()
|
||||
store.create_captcha("captcha-1", "ABCD", 120)
|
||||
|
||||
assert store.consume_captcha("captcha-1", "ABCD") is True
|
||||
assert store.consume_captcha("captcha-1", "ABCD") is False
|
||||
|
||||
|
||||
def test_admin_captcha_endpoint_returns_graphical_challenge():
|
||||
store = InMemoryAdminSessionStore()
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||||
|
||||
try:
|
||||
response = TestClient(app).get("/api/admin/auth/captcha")
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()["data"]
|
||||
assert data["captchaEnabled"] is True
|
||||
assert data["captchaId"]
|
||||
assert data["image"].startswith("data:image/svg+xml;base64,")
|
||||
assert data["expiresIn"] == 120
|
||||
|
||||
|
||||
def test_admin_access_token_contains_scoped_session_claims():
|
||||
user = AdminUser(
|
||||
id="admin-1",
|
||||
@@ -109,12 +135,20 @@ def test_admin_login_keeps_legacy_fields_and_sets_http_only_refresh_cookie():
|
||||
)
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: LoginDb(user)
|
||||
app.dependency_overrides[get_admin_session_store] = lambda: InMemoryAdminSessionStore()
|
||||
store = InMemoryAdminSessionStore()
|
||||
store.create_captcha("captcha-login", "ABCD", 120)
|
||||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/auth/login",
|
||||
json={"email": "admin@example.com", "password": "ChangeMe123!"},
|
||||
json={
|
||||
"email": "admin@example.com",
|
||||
"password": "ChangeMe123!",
|
||||
"captchaId": "captcha-login",
|
||||
"captchaCode": "ABCD",
|
||||
"rememberMe": False,
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
@@ -127,6 +161,7 @@ def test_admin_login_keeps_legacy_fields_and_sets_http_only_refresh_cookie():
|
||||
cookie = response.headers["set-cookie"]
|
||||
assert "HttpOnly" in cookie
|
||||
assert "Path=/api/admin/auth" in cookie
|
||||
assert "Max-Age=" not in cookie
|
||||
|
||||
|
||||
def test_refresh_rotates_access_token_and_logout_revokes_it():
|
||||
@@ -145,9 +180,16 @@ def test_refresh_rotates_access_token_and_logout_revokes_it():
|
||||
|
||||
try:
|
||||
client = TestClient(app)
|
||||
store.create_captcha("captcha-refresh", "ABCD", 120)
|
||||
login_response = client.post(
|
||||
"/api/admin/auth/login",
|
||||
json={"email": "admin@example.com", "password": "ChangeMe123!"},
|
||||
json={
|
||||
"email": "admin@example.com",
|
||||
"password": "ChangeMe123!",
|
||||
"captchaId": "captcha-refresh",
|
||||
"captchaCode": "ABCD",
|
||||
"rememberMe": False,
|
||||
},
|
||||
)
|
||||
old_token = login_response.json()["data"]["accessToken"]
|
||||
|
||||
@@ -155,6 +197,7 @@ def test_refresh_rotates_access_token_and_logout_revokes_it():
|
||||
new_token = refresh_response.json()["data"]["accessToken"]
|
||||
|
||||
assert refresh_response.status_code == 200
|
||||
assert "Max-Age=" not in refresh_response.headers["set-cookie"]
|
||||
assert new_token != old_token
|
||||
assert client.get("/api/admin/me", headers={"Authorization": f"Bearer {old_token}"}).status_code == 401
|
||||
assert client.get("/api/admin/me", headers={"Authorization": f"Bearer {new_token}"}).status_code == 200
|
||||
@@ -166,6 +209,45 @@ def test_refresh_rotates_access_token_and_logout_revokes_it():
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def test_remember_me_keeps_persistent_cookie_after_refresh_rotation():
|
||||
user = AdminUser(
|
||||
id="admin-1",
|
||||
email="admin@example.com",
|
||||
name="Admin",
|
||||
role="admin",
|
||||
passwordHash=hash_password("ChangeMe123!", rounds=4),
|
||||
isActive=True,
|
||||
)
|
||||
store = InMemoryAdminSessionStore()
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: LoginDb(user)
|
||||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||||
|
||||
try:
|
||||
client = TestClient(app)
|
||||
store.create_captcha("captcha-remember", "ABCD", 120)
|
||||
login_response = client.post(
|
||||
"/api/admin/auth/login",
|
||||
json={
|
||||
"email": "admin@example.com",
|
||||
"password": "ChangeMe123!",
|
||||
"captchaId": "captcha-remember",
|
||||
"captchaCode": "ABCD",
|
||||
"rememberMe": True,
|
||||
},
|
||||
)
|
||||
|
||||
assert login_response.status_code == 200
|
||||
assert "Max-Age=" in login_response.headers["set-cookie"]
|
||||
|
||||
refresh_response = client.post("/api/admin/auth/refresh")
|
||||
|
||||
assert refresh_response.status_code == 200
|
||||
assert "Max-Age=" in refresh_response.headers["set-cookie"]
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
def test_rbac_models_define_normalized_association_tables():
|
||||
assert AdminRole.__tablename__ == "AdminRole"
|
||||
assert AdminMenu.__tablename__ == "AdminMenu"
|
||||
|
||||
@@ -11,6 +11,7 @@ from app.models import AdminUser, Customer, HeroSlide, Lead, MediaAsset, Vehicle
|
||||
from app.routers import admin as admin_router
|
||||
from app.routers import public as public_router
|
||||
from app.routers.shared import site_config
|
||||
from app.redis_session import InMemoryAdminSessionStore, get_admin_session_store
|
||||
from app.schemas import LeadCreateIn, LeadQuery
|
||||
|
||||
|
||||
@@ -264,6 +265,39 @@ def test_admin_dynamic_routers_requires_admin_authentication():
|
||||
assert response.json() == {"code": 401, "msg": "请先登录后台", "data": None}
|
||||
|
||||
|
||||
def test_admin_login_requires_valid_captcha():
|
||||
admin_user = AdminUser(
|
||||
id="admin-test",
|
||||
email="admin@example.com",
|
||||
name="Admin",
|
||||
role="admin",
|
||||
passwordHash=hash_password("ChangeMe123!", rounds=4),
|
||||
isActive=True,
|
||||
)
|
||||
fake_db = FakeDb(scalar_values=[admin_user])
|
||||
store = InMemoryAdminSessionStore()
|
||||
store.create_captcha("captcha-test", "ABCD", 120)
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/auth/login",
|
||||
json={
|
||||
"email": "admin@example.com",
|
||||
"password": "ChangeMe123!",
|
||||
"captchaId": "captcha-test",
|
||||
"captchaCode": "WRONG",
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
assert response.status_code == 401
|
||||
assert response.json()["data"] is None
|
||||
|
||||
|
||||
def test_admin_login_returns_token_and_user():
|
||||
admin_user = AdminUser(
|
||||
id="admin-test",
|
||||
@@ -274,13 +308,22 @@ def test_admin_login_returns_token_and_user():
|
||||
isActive=True,
|
||||
)
|
||||
fake_db = FakeDb(scalar_values=[admin_user])
|
||||
store = InMemoryAdminSessionStore()
|
||||
store.create_captcha("captcha-contract", "ABCD", 120)
|
||||
app = create_app()
|
||||
app.dependency_overrides[get_db] = lambda: fake_db
|
||||
app.dependency_overrides[get_admin_session_store] = lambda: store
|
||||
|
||||
try:
|
||||
response = TestClient(app).post(
|
||||
"/api/admin/auth/login",
|
||||
json={"email": "admin@example.com", "password": "ChangeMe123!"},
|
||||
json={
|
||||
"email": "admin@example.com",
|
||||
"password": "ChangeMe123!",
|
||||
"captchaId": "captcha-contract",
|
||||
"captchaCode": "ABCD",
|
||||
"rememberMe": True,
|
||||
},
|
||||
)
|
||||
finally:
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
@@ -0,0 +1,93 @@
|
||||
from importlib.util import module_from_spec, spec_from_file_location
|
||||
from pathlib import Path
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
VERSIONS_DIR = Path(__file__).parents[1] / "alembic" / "versions"
|
||||
|
||||
|
||||
def load_migration():
|
||||
path = VERSIONS_DIR / "0034_media_assets.py"
|
||||
spec = spec_from_file_location("media_asset_migration", path)
|
||||
assert spec and spec.loader
|
||||
migration = module_from_spec(spec)
|
||||
spec.loader.exec_module(migration)
|
||||
return migration
|
||||
|
||||
|
||||
class Inspector:
|
||||
def __init__(self, tables: set[str]):
|
||||
self.tables = tables
|
||||
|
||||
def get_table_names(self) -> list[str]:
|
||||
return sorted(self.tables)
|
||||
|
||||
|
||||
def test_media_asset_migration_creates_the_upload_table_with_scope_columns(monkeypatch):
|
||||
migration = load_migration()
|
||||
created: dict[str, object] = {}
|
||||
indexes: list[tuple[object, ...]] = []
|
||||
|
||||
monkeypatch.setattr(migration, "inspect", lambda _bind: Inspector({"AdminUser", "AdminDepartment"}))
|
||||
monkeypatch.setattr(migration.op, "get_bind", lambda: object())
|
||||
monkeypatch.setattr(
|
||||
migration.op,
|
||||
"create_table",
|
||||
lambda *args, **kwargs: created.update(args=args, kwargs=kwargs),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
migration.op,
|
||||
"create_index",
|
||||
lambda *args, **kwargs: indexes.append((*args, *kwargs.values())),
|
||||
)
|
||||
|
||||
migration.upgrade()
|
||||
|
||||
assert migration.down_revision == "0033_remove_site_versions"
|
||||
assert created["args"][0] == "MediaAsset"
|
||||
columns = {column.name: column for column in created["args"][1:] if isinstance(column, sa.Column)}
|
||||
assert set(columns) == {
|
||||
"id",
|
||||
"url",
|
||||
"name",
|
||||
"mimeType",
|
||||
"sizeBytes",
|
||||
"group",
|
||||
"deptId",
|
||||
"createdById",
|
||||
"createdAt",
|
||||
"updatedAt",
|
||||
}
|
||||
assert columns["url"].unique
|
||||
assert columns["deptId"].nullable is False
|
||||
assert columns["deptId"].server_default.arg.text.strip("'") == migration.DEFAULT_DEPT_ID
|
||||
assert columns["createdById"].nullable is True
|
||||
assert {index[0] for index in indexes} == {
|
||||
"ix_MediaAsset_deptId",
|
||||
"ix_MediaAsset_createdById",
|
||||
}
|
||||
constraints = [constraint for constraint in created["args"][1:] if isinstance(constraint, sa.ForeignKeyConstraint)]
|
||||
primary_keys = [constraint for constraint in created["args"][1:] if isinstance(constraint, sa.PrimaryKeyConstraint)]
|
||||
assert primary_keys[0]._pending_colargs == ["id"]
|
||||
assert {
|
||||
(constraint.name, tuple(constraint.column_keys), tuple(constraint.elements[0].target_fullname.split(".")))
|
||||
for constraint in constraints
|
||||
} == {
|
||||
("fk_MediaAsset_deptId", ("deptId",), ("AdminDepartment", "id")),
|
||||
("fk_MediaAsset_createdById", ("createdById",), ("AdminUser", "id")),
|
||||
}
|
||||
|
||||
|
||||
def test_media_asset_migration_is_safe_when_table_already_exists(monkeypatch):
|
||||
migration = load_migration()
|
||||
calls: list[tuple[object, ...]] = []
|
||||
|
||||
monkeypatch.setattr(migration, "inspect", lambda _bind: Inspector({"MediaAsset"}))
|
||||
monkeypatch.setattr(migration.op, "get_bind", lambda: object())
|
||||
monkeypatch.setattr(migration.op, "create_table", lambda *args, **_kwargs: calls.append(args))
|
||||
monkeypatch.setattr(migration.op, "create_index", lambda *args, **_kwargs: calls.append(args))
|
||||
|
||||
migration.upgrade()
|
||||
|
||||
assert calls == []
|
||||
@@ -0,0 +1,82 @@
|
||||
from base64 import b64encode
|
||||
from hashlib import sha1
|
||||
from hmac import new as hmac_new
|
||||
from types import SimpleNamespace
|
||||
from urllib.parse import parse_qs, urlsplit
|
||||
|
||||
from app import media_urls
|
||||
|
||||
|
||||
def test_sign_oss_get_url_uses_oss_v1_query_signature():
|
||||
url = "https://one-feel-ota-data.oss-cn-guangzhou.aliyuncs.com/admin/site-config/image.webp"
|
||||
|
||||
signed = media_urls.sign_oss_get_url(
|
||||
url,
|
||||
access_key_id="test-access-key",
|
||||
access_key_secret="test-access-secret",
|
||||
endpoint="oss-cn-guangzhou.aliyuncs.com",
|
||||
bucket="one-feel-ota-data",
|
||||
expires_at=1_800_000_000,
|
||||
)
|
||||
|
||||
parsed = urlsplit(signed)
|
||||
query = parse_qs(parsed.query)
|
||||
string_to_sign = "GET\n\n\n1800000000\n/one-feel-ota-data/admin/site-config/image.webp"
|
||||
expected_signature = b64encode(
|
||||
hmac_new(b"test-access-secret", string_to_sign.encode("utf-8"), sha1).digest()
|
||||
).decode("ascii")
|
||||
|
||||
assert parsed.path == "/admin/site-config/image.webp"
|
||||
assert query["OSSAccessKeyId"] == ["test-access-key"]
|
||||
assert query["Expires"] == ["1800000000"]
|
||||
assert query["Signature"] == [expected_signature]
|
||||
|
||||
|
||||
def test_resolve_media_url_only_signs_the_configured_oss_host(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
media_urls,
|
||||
"get_settings",
|
||||
lambda: SimpleNamespace(
|
||||
oss_access_key_id="test-access-key",
|
||||
oss_access_key_secret="test-access-secret",
|
||||
oss_endpoint="oss-cn-guangzhou.aliyuncs.com",
|
||||
oss_bucket_name="one-feel-ota-data",
|
||||
),
|
||||
)
|
||||
monkeypatch.setattr(media_urls, "time", lambda: 1_800_000_000)
|
||||
|
||||
oss_url = "https://one-feel-ota-data.oss-cn-guangzhou.aliyuncs.com/admin/image.webp"
|
||||
external_url = "https://cdn.example.test/image.webp"
|
||||
|
||||
signed = media_urls.resolve_media_url(oss_url)
|
||||
|
||||
assert signed != oss_url
|
||||
assert "OSSAccessKeyId=test-access-key" in signed
|
||||
assert media_urls.resolve_media_url(external_url) == external_url
|
||||
|
||||
|
||||
def test_resolve_media_fields_signs_common_image_fields(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
"app.serializers.resolve_media_url",
|
||||
lambda value: f"signed:{value}" if value else value,
|
||||
)
|
||||
|
||||
from app.serializers import resolve_media_fields
|
||||
|
||||
result = resolve_media_fields(
|
||||
{
|
||||
"image": "oss-image",
|
||||
"avatar": "oss-avatar",
|
||||
"qrImage": "oss-qr",
|
||||
"gallery": ["oss-gallery-1", "oss-gallery-2"],
|
||||
"label": "unchanged",
|
||||
}
|
||||
)
|
||||
|
||||
assert result == {
|
||||
"image": "signed:oss-image",
|
||||
"avatar": "signed:oss-avatar",
|
||||
"qrImage": "signed:oss-qr",
|
||||
"gallery": ["signed:oss-gallery-1", "signed:oss-gallery-2"],
|
||||
"label": "unchanged",
|
||||
}
|
||||
Reference in new issue
Block a user