366 lines
14 KiB
Python
366 lines
14 KiB
Python
"""Walk the first-batch acceptance against the running API. Reads secrets from the environment."""
|
|
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import delete, select
|
|
|
|
load_dotenv()
|
|
|
|
from app.core.config import get_settings
|
|
from app.db import SessionLocal, get_engine
|
|
from app.models.rbac import AdminRole, AdminRoleResource, AdminUser, AdminUserRole
|
|
from app.models.security import AdminAuditLog, AdminIpBlacklist
|
|
from app.services.redis_client import get_redis
|
|
|
|
BASE = os.environ.get("ACCEPTANCE_BASE", "http://127.0.0.1:18000")
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise SystemExit(message)
|
|
|
|
|
|
def main() -> None:
|
|
get_settings()
|
|
username = os.environ["BOOTSTRAP_ADMIN_USERNAME"]
|
|
password = os.environ["BOOTSTRAP_ADMIN_PASSWORD"]
|
|
client = httpx.Client(base_url=BASE, timeout=30)
|
|
get_engine()
|
|
|
|
def captcha_code() -> tuple[str, str]:
|
|
issued = client.get("/api/admin/auth/captcha")
|
|
if issued.status_code != 200:
|
|
fail(f"captcha {issued.status_code}")
|
|
captcha_id = issued.json()["captchaId"]
|
|
stored = get_redis().get(f"gsfdc:captcha:{captcha_id}")
|
|
if not stored:
|
|
fail("captcha missing in redis")
|
|
return captcha_id, stored
|
|
|
|
def login(name: str, secret: str) -> httpx.Response:
|
|
captcha_id, code = captcha_code()
|
|
return client.post(
|
|
"/api/admin/auth/login",
|
|
json={"username": name, "password": secret, "captchaId": captcha_id, "captchaCode": code},
|
|
)
|
|
|
|
missing = client.get("/api/admin/users")
|
|
if missing.status_code != 401:
|
|
fail(f"expected 401, got {missing.status_code}")
|
|
|
|
captcha_id, code = captcha_code()
|
|
first = client.post(
|
|
"/api/admin/auth/login",
|
|
json={"username": username, "password": password, "captchaId": captcha_id, "captchaCode": code},
|
|
)
|
|
if first.status_code != 200:
|
|
fail(f"admin login {first.status_code} {first.text[:200]}")
|
|
second = client.post(
|
|
"/api/admin/auth/login",
|
|
json={"username": username, "password": password, "captchaId": captcha_id, "captchaCode": code},
|
|
)
|
|
if second.status_code != 401 or second.json().get("detail") != "用户名或密码错误":
|
|
fail(f"captcha reuse {second.status_code} {second.text[:200]}")
|
|
|
|
no_captcha = client.post(
|
|
"/api/admin/auth/login",
|
|
json={"username": username, "password": password, "captchaId": "", "captchaCode": ""},
|
|
)
|
|
if no_captcha.status_code != 401:
|
|
fail("login without captcha was accepted")
|
|
|
|
admin = login(username, password)
|
|
if admin.status_code != 200:
|
|
fail(f"admin login retry {admin.status_code}")
|
|
body = admin.json()
|
|
token = body["token"]
|
|
titles = [node["title"] for node in body["menus"]]
|
|
if "系统管理" not in titles:
|
|
fail(f"menus {titles}")
|
|
child_titles = {item["title"] for item in body["menus"][0]["children"]}
|
|
expected = {"用户管理", "角色管理", "菜单管理", "操作日志", "登录安全", "附件", "站点配置", "敏感词"}
|
|
if not expected <= child_titles:
|
|
fail(f"missing pages {expected - child_titles}")
|
|
if any(node["type"] == "button" for node in body["menus"]):
|
|
fail("button leaked into login menu")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
resources = client.get("/api/admin/resources", headers=headers)
|
|
if resources.status_code != 200:
|
|
fail(f"resources {resources.status_code}")
|
|
tree = resources.json()
|
|
|
|
def walk(nodes):
|
|
for node in nodes:
|
|
yield node
|
|
yield from walk(node.get("children") or [])
|
|
|
|
by_name = {node["name"]: node for node in walk(tree)}
|
|
system_id = by_name["System"]["id"]
|
|
user_page = by_name["User"]["id"]
|
|
add_button = next(node["id"] for node in walk(tree) if node.get("permission") == "user:add")
|
|
|
|
role_code = f"R_VIEW_{uuid.uuid4().hex[:8]}"
|
|
created_role = client.post(
|
|
"/api/admin/roles",
|
|
headers=headers,
|
|
json={"code": role_code, "name": "内容查看", "description": "验收"},
|
|
)
|
|
if created_role.status_code != 200:
|
|
fail(f"create role {created_role.status_code} {created_role.text[:200]}")
|
|
role_id = created_role.json()["id"]
|
|
granted = client.put(
|
|
f"/api/admin/roles/{role_id}/resources",
|
|
headers=headers,
|
|
json={"resourceIds": [system_id, user_page]},
|
|
)
|
|
if granted.status_code != 200:
|
|
fail(f"grant {granted.status_code} {granted.text[:200]}")
|
|
|
|
viewer_name = f"viewer_{uuid.uuid4().hex[:8]}"
|
|
viewer_password = "View1234"
|
|
created_user = client.post(
|
|
"/api/admin/users",
|
|
headers=headers,
|
|
json={
|
|
"username": viewer_name,
|
|
"password": viewer_password,
|
|
"displayName": "内容查看员",
|
|
"mobile": "13800138000",
|
|
"roleIds": [role_id],
|
|
},
|
|
)
|
|
if created_user.status_code != 200:
|
|
fail(f"create user {created_user.status_code} {created_user.text[:300]}")
|
|
created_body = created_user.json()
|
|
if created_body.get("mobileMask") != "138****8000":
|
|
fail(f"mask {created_body}")
|
|
if "password" in created_body or "mobileCipher" in created_body:
|
|
fail("secret leaked in user response")
|
|
viewer_id = created_body["id"]
|
|
|
|
viewer_login = login(viewer_name, viewer_password)
|
|
if viewer_login.status_code != 200:
|
|
fail(f"viewer login {viewer_login.status_code} {viewer_login.text[:200]}")
|
|
viewer_body = viewer_login.json()
|
|
pages = [node["title"] for node in viewer_body["menus"][0]["children"]]
|
|
if pages != ["用户管理"]:
|
|
fail(f"viewer pages {pages}")
|
|
if viewer_body["permissions"]:
|
|
fail(f"viewer permissions {viewer_body['permissions']}")
|
|
|
|
granted_add = client.put(
|
|
f"/api/admin/roles/{role_id}/resources",
|
|
headers=headers,
|
|
json={"resourceIds": [system_id, user_page, add_button]},
|
|
)
|
|
if granted_add.status_code != 200:
|
|
fail(f"grant add {granted_add.status_code}")
|
|
viewer_again = login(viewer_name, viewer_password)
|
|
if viewer_again.json().get("permissions") != ["user:add"]:
|
|
fail(f"permissions after grant {viewer_again.json().get('permissions')}")
|
|
|
|
disabled = client.put(
|
|
f"/api/admin/users/{viewer_id}",
|
|
headers=headers,
|
|
json={"displayName": "内容查看员", "mobile": "13800138000", "status": "disabled", "roleIds": [role_id]},
|
|
)
|
|
if disabled.status_code != 200:
|
|
fail(f"disable {disabled.status_code} {disabled.text[:200]}")
|
|
if login(viewer_name, viewer_password).status_code != 401:
|
|
fail("disabled user logged in")
|
|
enabled = client.put(
|
|
f"/api/admin/users/{viewer_id}",
|
|
headers=headers,
|
|
json={"displayName": "内容查看员", "status": "active", "roleIds": [role_id]},
|
|
)
|
|
if enabled.status_code != 200 or login(viewer_name, viewer_password).status_code != 200:
|
|
fail("reenabled user could not login")
|
|
cancelled = client.put(
|
|
f"/api/admin/users/{viewer_id}",
|
|
headers=headers,
|
|
json={"displayName": "内容查看员", "status": "cancelled", "roleIds": [role_id]},
|
|
)
|
|
if cancelled.status_code != 200:
|
|
fail(f"cancel {cancelled.status_code} {cancelled.text[:200]}")
|
|
if login(viewer_name, viewer_password).status_code != 401:
|
|
fail("cancelled user logged in")
|
|
listed = client.get("/api/admin/users", headers=headers, params={"username": viewer_name})
|
|
if listed.status_code != 200 or not listed.json()["records"]:
|
|
fail("cancelled user missing from list")
|
|
|
|
weak = client.post(
|
|
"/api/admin/users",
|
|
headers=headers,
|
|
json={"username": f"weak_{uuid.uuid4().hex[:6]}", "password": "abcdefgh", "displayName": "弱", "roleIds": []},
|
|
)
|
|
if weak.status_code != 400:
|
|
fail("weak password accepted")
|
|
|
|
policy = client.put(
|
|
"/api/admin/login-security",
|
|
headers=headers,
|
|
json={"loginMaxFailures": 2, "loginLockMinutes": 15},
|
|
)
|
|
if policy.status_code != 200:
|
|
fail(f"policy {policy.status_code} {policy.text[:200]}")
|
|
lock_name = f"lock_{uuid.uuid4().hex[:6]}"
|
|
lock_password = "Lock1234"
|
|
lock_user = client.post(
|
|
"/api/admin/users",
|
|
headers=headers,
|
|
json={"username": lock_name, "password": lock_password, "displayName": "锁定", "roleIds": []},
|
|
)
|
|
if lock_user.status_code != 200:
|
|
fail(f"lock user {lock_user.status_code}")
|
|
for _ in range(2):
|
|
bad_id, _bad_code = captcha_code()
|
|
client.post(
|
|
"/api/admin/auth/login",
|
|
json={"username": lock_name, "password": "Wrong123", "captchaId": bad_id, "captchaCode": "ZZZZ"},
|
|
)
|
|
locked = login(lock_name, lock_password)
|
|
if locked.status_code != 401:
|
|
fail("locked account accepted the right password")
|
|
db = SessionLocal()
|
|
try:
|
|
row = db.scalar(select(AdminUser).where(AdminUser.username == lock_name))
|
|
row.locked_until = datetime.now(timezone.utc) - timedelta(seconds=5)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
if login(lock_name, lock_password).status_code != 200:
|
|
fail("login failed after lock expired")
|
|
|
|
client.put(
|
|
"/api/admin/login-security",
|
|
headers=headers,
|
|
json={"loginMaxFailures": 5, "loginLockMinutes": 15},
|
|
)
|
|
|
|
blocked_ip = "127.0.0.1"
|
|
added = client.post("/api/admin/ip-blacklist", headers=headers, json={"ip": blocked_ip, "remark": "acceptance"})
|
|
if added.status_code != 200:
|
|
fail(f"blacklist add {added.status_code} {added.text[:200]}")
|
|
blacklist_id = added.json()["id"]
|
|
try:
|
|
blocked = client.get("/api/admin/users", headers=headers)
|
|
if blocked.status_code != 403:
|
|
fail(f"blacklisted request was not rejected: {blocked.status_code}")
|
|
finally:
|
|
db = SessionLocal()
|
|
try:
|
|
db.execute(delete(AdminIpBlacklist).where(AdminIpBlacklist.id == blacklist_id))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
restored = client.get("/api/admin/users", headers=headers)
|
|
if restored.status_code != 200:
|
|
fail(f"request still blocked after removal: {restored.status_code}")
|
|
|
|
logs = client.get("/api/admin/audit-logs", headers=headers, params={"username": username, "current": 1, "size": 20})
|
|
if logs.status_code != 200 or logs.json()["total"] < 1:
|
|
fail(f"audit {logs.status_code} {logs.text[:200]}")
|
|
raw_logs = logs.text
|
|
if password in raw_logs or "captchaCode" in raw_logs:
|
|
fail("audit response contains a secret")
|
|
|
|
over = client.post(
|
|
"/api/admin/files",
|
|
headers=headers,
|
|
data={"bucket": "public"},
|
|
files={"file": ("big.bin", b"x" * (10 * 1024 * 1024 + 1), "application/octet-stream")},
|
|
)
|
|
if over.status_code != 400:
|
|
fail(f"oversize {over.status_code}")
|
|
public = client.post(
|
|
"/api/admin/files",
|
|
headers=headers,
|
|
data={"bucket": "public"},
|
|
files={"file": ("ok.txt", b"hello", "text/plain")},
|
|
)
|
|
if public.status_code != 200:
|
|
fail(f"public upload {public.status_code} {public.text[:200]}")
|
|
public_url = public.json()["url"]
|
|
fetched = httpx.get(public_url, timeout=20)
|
|
if fetched.status_code != 200 or fetched.content != b"hello":
|
|
fail(f"public fetch {fetched.status_code}")
|
|
private = client.post(
|
|
"/api/admin/files",
|
|
headers=headers,
|
|
data={"bucket": "private"},
|
|
files={"file": ("secret.txt", b"secret", "text/plain")},
|
|
)
|
|
if private.status_code != 200 or not private.json().get("expiresAt"):
|
|
fail(f"private upload {private.status_code} {private.text[:200]}")
|
|
|
|
site = client.put(
|
|
"/api/admin/site-settings",
|
|
headers=headers,
|
|
json={
|
|
"site_name": "验收站点",
|
|
"icp_text": "陇ICP备19001227号",
|
|
"link_cirea": "https://www.cirea.org.cn/",
|
|
},
|
|
)
|
|
if site.status_code != 200:
|
|
fail(f"site {site.status_code} {site.text[:200]}")
|
|
public_site = client.get("/api/site/settings")
|
|
public_body = public_site.json()
|
|
if public_body.get("siteName") != "验收站点" or public_body.get("linkCirea") != "https://www.cirea.org.cn/":
|
|
fail(f"public site {public_body}")
|
|
if "loginMaxFailures" in public_body or "login_max_failures" in public_body:
|
|
fail("public site leaked lock settings")
|
|
|
|
probe_word = f"验收敏感词{uuid.uuid4().hex[:6]}"
|
|
word = client.post("/api/admin/sensitive-words", headers=headers, json={"word": probe_word})
|
|
if word.status_code != 200:
|
|
fail(f"word {word.status_code} {word.text[:200]}")
|
|
preview = client.post(
|
|
"/api/admin/content/preview-filter",
|
|
headers=headers,
|
|
json={"html": f'<p onclick="alert(1)">{probe_word}</p><script>alert(1)</script>'},
|
|
)
|
|
if preview.status_code != 200:
|
|
fail(f"preview {preview.status_code}")
|
|
preview_body = preview.json()
|
|
if "script" in preview_body["html"].lower() or "onclick" in preview_body["html"].lower():
|
|
fail(f"filter kept active content {preview_body['html']}")
|
|
if probe_word not in preview_body["hits"]:
|
|
fail(f"hits {preview_body['hits']}")
|
|
articles = client.get("/api/site/settings")
|
|
if probe_word in articles.text:
|
|
fail("preview became public content")
|
|
client.delete(f"/api/admin/sensitive-words/{word.json()['id']}", headers=headers)
|
|
client.delete(f"/api/admin/files/{public.json()['id']}", headers=headers)
|
|
client.delete(f"/api/admin/files/{private.json()['id']}", headers=headers)
|
|
client.put(
|
|
"/api/admin/site-settings",
|
|
headers=headers,
|
|
json={"site_name": "甘肃省房地产价格评估行业协会", "link_cirea": ""},
|
|
)
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
user_ids = [
|
|
row.id
|
|
for row in db.scalars(select(AdminUser).where(AdminUser.username.in_([viewer_name, lock_name]))).all()
|
|
]
|
|
if user_ids:
|
|
db.execute(delete(AdminUserRole).where(AdminUserRole.user_id.in_(user_ids)))
|
|
db.execute(delete(AdminUser).where(AdminUser.id.in_(user_ids)))
|
|
db.execute(delete(AdminRoleResource).where(AdminRoleResource.role_id == role_id))
|
|
db.execute(delete(AdminRole).where(AdminRole.id == role_id))
|
|
db.execute(delete(AdminIpBlacklist).where(AdminIpBlacklist.ip == blocked_ip))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
print("live acceptance passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|