""" Code Audit Script ================== Kiểm tra toàn bộ code tuân thủ project rules. Chạy: python audit.py """ import os, re, sqlite3, json, time from datetime import datetime BASE_DIR = os.path.dirname(__file__) RESULTS = [] def check(rule, desc, passed, detail=""): status = "✅ PASS" if passed else "❌ FAIL" RESULTS.append((rule, desc, status, detail)) print(f" {status} [{rule}] {desc}" + (f" — {detail}" if detail else "")) def audit(): print("=" * 60) print("CODE AUDIT — Hikvision Attendance Project") print(f"Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("=" * 60) # === R1: Lock Prevention === print("\n🔒 R1: Lock Prevention") hk = open(os.path.join(BASE_DIR, 'hikvision.py'), 'r', encoding='utf-8').read() check("R1.1", "Max POST per session = 18", "MAX_POSTS_PER_SESSION" in hk and "post_count >= MAX_POSTS_PER_SESSION" in hk) check("R1.2", "Session cooldown (9s wait)", "SESSION_COOLDOWN_SECONDS" in hk or "time.sleep(SESSION_COOLDOWN_SECONDS)" in hk) check("R1.3", "Lock detection implemented", "'lock' in" in hk.lower() or "is_locked" in hk) check("R1.4", "Pre-sync lock check", "LOCK_CHECK_BEFORE_SYNC" in hk and "test_connection" in hk) check("R1.5", "Stop on 401", "401" in hk and "stop" in hk.lower()) check("R1.6", "Log session creation", "log_action" in hk and "SESSION" in hk) # === R2: Security === print("\n🔐 R2: Security") check("R2.1", "Password from config (not hardcoded)", "config['password']" in hk or "self.config" in hk, "Password loaded from config.json") # Check no password in log calls log_calls = re.findall(r'log_action\([^)]*password[^)]*\)', hk, re.IGNORECASE) check("R2.4", "No password in logs", len(log_calls) == 0, f"Found {len(log_calls)} log calls with 'password'" if log_calls else "Clean") # R2.2: .gitignore gitignore = os.path.join(BASE_DIR, '.gitignore') if os.path.exists(gitignore): gi = open(gitignore).read() check("R2.2", "config.json in .gitignore", "config.json" in gi) else: check("R2.2", ".gitignore exists", False, "No .gitignore file") # === R3: Data Integrity === print("\n💾 R3: Data Integrity") check("R3.1", "UNIQUE constraint on attendance", "UNIQUE(raw_time, employee_no)" in hk) # R3.2: Check for SQL DELETE/TRUNCATE statements (not comments) import re as _re sql_deletes = _re.findall(r'(?:c\.execute|cursor\.execute).*(?:DELETE|TRUNCATE)', hk, _re.IGNORECASE) check("R3.2", "No SQL DELETE/TRUNCATE statements", "INSERT OR IGNORE" in hk and len(sql_deletes) == 0, f"Found {len(sql_deletes)} SQL delete ops" if sql_deletes else "Clean - uses INSERT OR IGNORE") check("R3.3", "Sync log recorded", "sync_log" in hk and "_log_sync" in hk) check("R3.4", "Database indexes exist", "CREATE INDEX" in hk) # === R4: Logging === print("\n📝 R4: Logging") check("R4.1", "log_action used for sync", hk.count("log_action") >= 5, f"{hk.count('log_action')} calls") check("R4.2", "Performance logging", "log_performance" in hk or "@timed" in hk) log_file = os.path.join(BASE_DIR, 'logger.py') if os.path.exists(log_file): lgr = open(log_file, 'r', encoding='utf-8').read() check("R4.3", "Daily log rotation", "TimedRotatingFileHandler" in lgr and "midnight" in lgr) else: check("R4.3", "Logger module exists", False) # === R5: Performance === print("\n⚡ R5: Performance") check("R5.1", "Sync day timing (<3s)", "PERFORMANCE_WARN_SYNC_DAY_MS" in hk) check("R5.2", "Query timing (<1s)", "PERFORMANCE_WARN_QUERY_MS" in hk or "@timed" in hk) api_file = os.path.join(BASE_DIR, 'attendance_api.py') if os.path.exists(api_file): api = open(api_file, 'r', encoding='utf-8').read() check("R5.3", "API response timing (<2s)", "PERFORMANCE_WARN_API_MS" in api or "log_performance" in api) else: check("R5.3", "API module exists", False) # === R6: Excel Safety === print("\n📊 R6: Excel Safety") if os.path.exists(api_file): api = open(api_file, 'r', encoding='utf-8').read() # R6.1: GET handlers should use get_attendance/get_employees, not HikvisionDevice directly get_section = api.split("def api_sync")[0] if "def api_sync" in api else api # Only flag if GET handler functions create HikvisionDevice get_funcs = _re.findall(r'def api_\w+.*?(?=def |$)', get_section, _re.DOTALL) device_in_get = [f for f in get_funcs if 'HikvisionDevice' in f and 'test' not in f] check("R6.1", "GET endpoints don't create HikvisionDevice", len(device_in_get) == 0, "GET endpoints read from cache only" if not device_in_get else f"{len(device_in_get)} GET handlers touch device") check("R6.3", "Refresh is cache-only", "get_attendance" in api and "get_employees" in api) check("R6.4", "JSON format responses", api.count("jsonify") >= 5) # === Database Check === print("\n🗄️ Database Check") db_path = os.path.join(BASE_DIR, 'data', 'attendance.db') if os.path.exists(db_path): conn = sqlite3.connect(db_path) c = conn.cursor() c.execute("SELECT COUNT(*) FROM attendance") total = c.fetchone()[0] c.execute("SELECT COUNT(DISTINCT employee_no) FROM attendance") emps = c.fetchone()[0] c.execute("SELECT MIN(date), MAX(date) FROM attendance") d = c.fetchone() conn.close() check("DB", f"Database has data", total > 0, f"{total} records, {emps} employees, {d[0]}→{d[1]}") else: check("DB", "Database exists", False, db_path) # === Summary === passed = sum(1 for r in RESULTS if "PASS" in r[2]) failed = sum(1 for r in RESULTS if "FAIL" in r[2]) print("\n" + "=" * 60) print(f"AUDIT RESULT: {passed} PASSED, {failed} FAILED / {len(RESULTS)} total") print("=" * 60) # Write audit report report_path = os.path.join(BASE_DIR, 'logs', f'audit_{datetime.now().strftime("%Y%m%d_%H%M%S")}.txt') os.makedirs(os.path.dirname(report_path), exist_ok=True) with open(report_path, 'w', encoding='utf-8') as f: f.write(f"AUDIT REPORT — {datetime.now().isoformat()}\n") f.write("=" * 60 + "\n") for rule, desc, status, detail in RESULTS: f.write(f"{status} [{rule}] {desc}") if detail: f.write(f" — {detail}") f.write("\n") f.write(f"\nTotal: {passed} PASSED, {failed} FAILED / {len(RESULTS)}\n") print(f"\n📄 Report saved: {report_path}") return failed == 0 if __name__ == '__main__': audit()