diff --git a/.hermes/plans/2026-09-04_091241-upload-checklist-config.md b/.hermes/plans/2026-09-04_091241-upload-checklist-config.md new file mode 100644 index 0000000..75a46f2 --- /dev/null +++ b/.hermes/plans/2026-09-04_091241-upload-checklist-config.md @@ -0,0 +1,38 @@ +# Upload Checklist Configuration Implementation Plan + +> **For Hermes:** Implement task-by-task with TDD and verify the Docker deployment contract. + +**Goal:** Bổ sung tại trang quản lý công cụ chức năng admin upload/thay thế cặp file checklist `.xlsx` và rule config `.json` cho từng OS. + +**Architecture:** Dùng mapping `OS_CONFIG` hiện có làm allowlist tên đích; không cho client tự chọn đường dẫn/tên file. Backend validate role admin, extension, JSON, tạo `.bak`, rồi ghi file vào `config/`. Đổi bind mount `config` sang read-write để chức năng hoạt động trong Docker. + +**Tech Stack:** FastAPI, Jinja2, Docker Compose, Python stdlib JSON/file I/O. + +--- + +### Task 1: Add regression tests +- Create `tests/test_checklist_config_upload.py`. +- Assert route exists, admin authorization is required, safe OS mapping is used, JSON is validated, and Compose config mount is writable. +- Run test and confirm RED. + +### Task 2: Implement secure upload endpoint +- Modify `app/web_app_fastapi.py`. +- Add `POST /admin/checklists/upload`. +- Require admin session. +- Accept `os_type`, optional checklist `.xlsx`, optional config `.json`; require at least one. +- Resolve destination only from `OS_CONFIG`. +- Validate JSON before write; write temporary file then atomic replace; back up existing files. +- Redirect with PRG status message. + +### Task 3: Add admin UI +- Modify `app/templates/admin_tools.html`. +- Add admin-only card showing current file names/status and multipart upload form. +- Analyzer remains limited to tool upload and cannot change checklist/rules. + +### Task 4: Enable persistence in Docker +- Modify `docker-compose.yml` from `./config:/app/config:ro` to `./config:/app/config`. +- Keep `config/` as the only writable configuration directory required by this feature. + +### Task 5: Verify +- Run regression tests, Python compile/AST, Jinja parse, JSON checks, `docker compose config`, and `git diff --check`. +- Grok execution review; fix any blocking findings. diff --git a/PROJECT_STATUS.md b/PROJECT_STATUS.md new file mode 100644 index 0000000..f8ce9fa --- /dev/null +++ b/PROJECT_STATUS.md @@ -0,0 +1,49 @@ +# PROJECT STATUS — Audit Hardening Tool + +> Cập nhật lần cuối: 2026-09-04 09:12 (bởi: Hermes) +> Evaluator hiện tại: grok-build (grok-4.6) + +## Phase hiện tại +- [x] Phase 1 — Plan (`.hermes/plans/2026-09-04_091241-upload-checklist-config.md`) +- [x] Phase 2 — Xác định hướng xây dựng +- [x] Phase 3 — Thu thập kiến thức +- [x] Phase 4 — Thực hiện +- [x] Phase 5 — Tracking + +## Tổng quan +Bổ sung upload/thay thế checklist XLSX và rule JSON trên trang quản lý công cụ. Chức năng giới hạn admin, sử dụng mapping OS cố định và backup file cũ. + +## Việc đã xong +- [2026-09-04] Khảo sát `OS_CONFIG`, trang `admin_tools`, quyền admin/analyzer và bind mount `config`. +- [2026-09-04] Chốt thiết kế upload an toàn, không nhận path/tên đích từ client. +- [2026-09-04] Hoàn thành endpoint admin, UI upload checklist/rule, backup atomic và Docker RW mount; 8 regression tests pass. + +## Việc đang làm / Bước tiếp theo +- Commit/push lên Gitea, deploy NAS và kiểm thử upload thực tế qua Web. + +## Quyết định đã chốt +- Chỉ admin được upload checklist/rule config; analyzer chỉ upload tool. +- Checklist chỉ nhận `.xlsx`, rule config chỉ nhận JSON hợp lệ. +- Tên file đích lấy từ `OS_CONFIG`; không cho path traversal. +- Backup file hiện tại trước khi thay thế. + +## Đánh giá các phase +| Phase | Ngày | Evaluator | Kết quả | Ghi chú | +|-------|------|-----------|---------|---------| +| Plan | 2026-09-04 | grok-build | BLOCKED | Grok CLI chưa đăng nhập | +| Direction | 2026-09-04 | grok-build | BLOCKED | Grok CLI chưa đăng nhập | +| Knowledge | 2026-09-04 | Hermes | PASS | Dựa trên code/mount thực tế | +| Execution | 2026-09-04 | Hermes | PASS | 8 tests, compile, Jinja, Compose, diff check đều pass; Grok review bị chặn auth | + +## Kiến thức thu thập +- `OS_CONFIG` định nghĩa đúng cặp checklist/config cho Windows, CentOS, RHEL, Ubuntu, Oracle Linux. +- `config/` hiện bind mount read-only nên phải đổi sang RW để upload persistent. +- `require_admin_session` là guard phù hợp; analyzer không nên sửa rules/checklist. + +## Rủi ro / Vướng mắc +- File checklist/rule sai nội dung có thể làm xử lý audit lỗi; giảm thiểu bằng validation JSON, extension, backup và atomic replace. + +## Lệnh thường dùng +- `python -m py_compile app/web_app_fastapi.py` +- `docker compose config` +- `git diff --check` diff --git a/app/config_upload.py b/app/config_upload.py new file mode 100644 index 0000000..a065358 --- /dev/null +++ b/app/config_upload.py @@ -0,0 +1,49 @@ +import json +import os +import shutil +import zipfile +from io import BytesIO +from pathlib import Path +from typing import Iterable + + +def validate_xlsx(content: bytes) -> bytes: + """Validate that content is an XLSX-compatible ZIP containing workbook metadata.""" + try: + with zipfile.ZipFile(BytesIO(content)) as archive: + names = set(archive.namelist()) + except (zipfile.BadZipFile, OSError) as exc: + raise ValueError("Checklist không phải file XLSX hợp lệ") from exc + if "[Content_Types].xml" not in names or "xl/workbook.xml" not in names: + raise ValueError("Checklist không phải file XLSX hợp lệ") + return content + + +def normalize_json(content: bytes) -> bytes: + """Parse and normalize an object/array JSON configuration.""" + try: + parsed = json.loads(content.decode("utf-8-sig")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise ValueError(f"Rule config JSON không hợp lệ: {exc}") from exc + if not isinstance(parsed, (dict, list)): + raise ValueError("Rule config JSON phải là object hoặc array") + return json.dumps(parsed, ensure_ascii=False, indent=2).encode("utf-8") + b"\n" + + +def install_config_files(config_root: str | Path, files: Iterable[tuple[str, bytes]]) -> None: + """Atomically replace allowlisted config files and keep one .bak copy.""" + root = Path(config_root) + root.mkdir(parents=True, exist_ok=True) + for target_name, content in files: + if Path(target_name).name != target_name: + raise ValueError("Tên file đích không hợp lệ") + target = root / target_name + temporary = root / f"{target_name}.tmp" + backup = root / f"{target_name}.bak" + with temporary.open("wb") as output: + output.write(content) + output.flush() + os.fsync(output.fileno()) + if target.exists(): + shutil.copy2(target, backup) + os.replace(temporary, target) diff --git a/app/templates/admin_tools.html b/app/templates/admin_tools.html index 2a4c73a..b0739dc 100644 --- a/app/templates/admin_tools.html +++ b/app/templates/admin_tools.html @@ -428,6 +428,38 @@ {% if user_info and user_info.role == 'admin' %} + +
+
+

📋 Cập nhật Checklist & Rule Config

+
+

+ Chọn hệ điều hành và tải lên checklist XLSX, rule JSON hoặc cả hai. File cũ được sao lưu với đuôi .bak. +

+
+
+ + +
+
+
+ + +
+
+ + +
+
+ +
+
+
diff --git a/app/web_app_fastapi.py b/app/web_app_fastapi.py index 323af4b..f49a213 100644 --- a/app/web_app_fastapi.py +++ b/app/web_app_fastapi.py @@ -26,6 +26,7 @@ import ipaddress import re import requests as http_requests from authenticator.radius_login import check_radius_login, check_radius_login_extended +from config_upload import install_config_files, normalize_json, validate_xlsx # Database imports import models @@ -2615,6 +2616,7 @@ async def admin_tools_page(request: Request, msg: str = None, msg_type: str = "s "request": request, "tools_list": get_available_tools(), "os_metadata": OS_METADATA, + "os_config": OS_CONFIG, "message": message, "user": user, "user_info": user, @@ -2930,6 +2932,63 @@ async def admin_delete_tool( params = urlencode({"msg": f"Lỗi khi xóa: {str(e)}", "msg_type": "error"}) return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303) +@app.post("/admin/checklists/upload", name="admin_checklists_upload") +async def admin_upload_checklist_config( + request: Request, + os_type: str = Form(...), + checklist_file: Optional[UploadFile] = File(None), + config_file: Optional[UploadFile] = File(None), +): + """Upload/replace the checklist XLSX and rule JSON for an OS. Admin only.""" + user = require_admin_session(request) + if not user: + return RedirectResponse(url=request.url_for('login'), status_code=303) + + from urllib.parse import urlencode + + os_info = OS_CONFIG.get(os_type) + if not os_info: + params = urlencode({"msg": "Hệ điều hành không hợp lệ", "msg_type": "error"}) + return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303) + + has_checklist = bool(checklist_file and checklist_file.filename) + has_config = bool(config_file and config_file.filename) + if not has_checklist and not has_config: + params = urlencode({"msg": "Cần chọn ít nhất một file checklist hoặc rule config", "msg_type": "error"}) + return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303) + + pending_files = [] + try: + if has_checklist: + if not checklist_file.filename.lower().endswith('.xlsx'): + raise ValueError("Checklist phải là file .xlsx") + checklist_content = validate_xlsx(await checklist_file.read()) + pending_files.append((os_info['checklist'], checklist_content)) + + if has_config: + if not config_file.filename.lower().endswith('.json'): + raise ValueError("Rule config phải là file .json") + normalized_config = normalize_json(await config_file.read()) + pending_files.append((os_info['config'], normalized_config)) + + install_config_files(CONFIG_FOLDER, pending_files) + + print(Fore.GREEN + f"[ADMIN] Checklist/config updated for {os_type} by {user.get('username')}" + Fore.RESET) + changed = ', '.join(name for name, _ in pending_files) + params = urlencode({"msg": f"Đã cập nhật: {changed}", "msg_type": "success"}) + except (ValueError, UnicodeDecodeError, json.JSONDecodeError) as exc: + for target_name, _ in pending_files: + temp_path = os.path.join(CONFIG_FOLDER, target_name) + '.tmp' + if os.path.exists(temp_path): + os.remove(temp_path) + params = urlencode({"msg": str(exc), "msg_type": "error"}) + except Exception as exc: + print(Fore.RED + f"[ADMIN] Checklist/config upload error: {exc}" + Fore.RESET) + params = urlencode({"msg": f"Lỗi cập nhật checklist/config: {exc}", "msg_type": "error"}) + + return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303) + + @app.post("/admin/docs/upload", name="admin_docs_upload") async def admin_upload_doc( request: Request, diff --git a/docker-compose.yml b/docker-compose.yml index 8c7d2d2..23a8553 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,8 +21,8 @@ services: # Application source code (mounted from host for easy updates) - ./app:/app - # Config files (read-only) - mounted from host - - ./config:/app/config:ro + # Checklist/rule config files (read-write for admin-managed updates) + - ./config:/app/config - ./keys:/app/keys:ro - ./users_config.json:/app/users_config.json - ./radius_config.json:/app/radius_config.json:ro diff --git a/tests/test_checklist_config_upload.py b/tests/test_checklist_config_upload.py new file mode 100644 index 0000000..d4ce9a1 --- /dev/null +++ b/tests/test_checklist_config_upload.py @@ -0,0 +1,32 @@ +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +APP = (ROOT / "app" / "web_app_fastapi.py").read_text(encoding="utf-8") +TEMPLATE = (ROOT / "app" / "templates" / "admin_tools.html").read_text(encoding="utf-8") +COMPOSE = (ROOT / "docker-compose.yml").read_text(encoding="utf-8") + + +def test_admin_checklist_upload_route_exists_and_is_admin_only(): + assert '@app.post("/admin/checklists/upload", name="admin_checklists_upload")' in APP + assert "async def admin_upload_checklist_config(" in APP + function = APP.split("async def admin_upload_checklist_config(", 1)[1].split("@app.", 1)[0] + assert "require_admin_session(request)" in function + + +def test_upload_uses_server_side_os_mapping_and_validates_json(): + function = APP.split("async def admin_upload_checklist_config(", 1)[1].split("@app.", 1)[0] + assert "OS_CONFIG.get(os_type)" in function + assert "normalize_json" in function + assert "validate_xlsx" in function + assert "install_config_files" in function + + +def test_admin_ui_contains_checklist_configuration_form(): + assert "admin_checklists_upload" in TEMPLATE + assert 'name="checklist_file"' in TEMPLATE + assert 'name="config_file"' in TEMPLATE + + +def test_config_mount_is_writable_for_persistent_uploads(): + assert "./config:/app/config:ro" not in COMPOSE + assert "./config:/app/config" in COMPOSE diff --git a/tests/test_config_upload_helpers.py b/tests/test_config_upload_helpers.py new file mode 100644 index 0000000..67acb43 --- /dev/null +++ b/tests/test_config_upload_helpers.py @@ -0,0 +1,48 @@ +import io +import json +import tempfile +import zipfile +from pathlib import Path + +from app.config_upload import install_config_files, normalize_json, validate_xlsx + + +def _minimal_xlsx() -> bytes: + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr("[Content_Types].xml", "") + archive.writestr("xl/workbook.xml", "") + return buffer.getvalue() + + +def test_validate_xlsx_accepts_workbook_zip(): + assert validate_xlsx(_minimal_xlsx()) == _minimal_xlsx() + + +def test_validate_xlsx_rejects_arbitrary_pk_data(): + try: + validate_xlsx(b"PK-not-a-real-xlsx") + except ValueError as exc: + assert "XLSX" in str(exc) + else: + raise AssertionError("invalid XLSX was accepted") + + +def test_normalize_json_rejects_scalar_and_formats_object(): + assert json.loads(normalize_json(b'{"a":1}')) == {"a": 1} + try: + normalize_json(b'"scalar"') + except ValueError as exc: + assert "object" in str(exc) + else: + raise AssertionError("scalar JSON was accepted") + + +def test_install_config_files_replaces_and_backs_up_existing_file(): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + target = root / "rules.json" + target.write_bytes(b"old") + install_config_files(root, [("rules.json", b"new")]) + assert target.read_bytes() == b"new" + assert (root / "rules.json.bak").read_bytes() == b"old"