同样使用 HTTPDigestAuth
import json
import binascii
import base64
import hashlib
import time
import requests
import argparse
import uuid
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad, pad
from requests.auth import HTTPDigestAuth# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
DEFAULT_IP = "192.168.1.192"
DEFAULT_PORT = "8080"
USERNAME = "admin"
PASSWORD = "password"# 认证对象
DIGEST_AUTH = HTTPDigestAuth(USERNAME, PASSWORD)# ---------------------------------------------------------------------------
# Key Derivation
# ---------------------------------------------------------------------------
def get_aes_key(host_str, username, password, auth, proxies=None):"""Derive AES Key dynamically from device capabilities."""try:url = f"http://{host_str}/ISAPI/Security/capabilities?username={username}"print(f"Fetching capabilities from {url}...")resp = requests.get(url, auth=auth, timeout=10, verify=False, proxies=proxies)resp.raise_for_status()root = ET.fromstring(resp.text)ns = {'ns': 'http://www.isapi.org/ver20/XMLSchema'}salt_node = root.find('ns:salt', ns)if salt_node is None:print("Error: Could not find <salt> in capabilities response.")return Nonesalt = salt_node.textprint(f"Got Salt: {salt}")# 1. Calculate IrreversibleKeycombined_string = f"{username}{salt}{password}"irreversible_key = hashlib.sha256(combined_string.encode('utf-8')).hexdigest()# 2. Calculate Final Key with Challenge# Note: The challenge "AaBbCcDd1234!@#$" appears to be hardcoded or specific to this auth modechallenge = "AaBbCcDd1234!@#$" combined_for_hash = f"{irreversible_key}{challenge}"key = hashlib.sha256(combined_for_hash.encode('utf-8')).hexdigest()# 3. Iterate hashingiterations = 100for _ in range(1, iterations):key = hashlib.sha256(key.encode()).hexdigest()aes_key = key[:32]print(f"Derived AES Key: {aes_key}")return aes_keyexcept Exception as e:print(f"Key derivation failed: {e}")return None# ---------------------------------------------------------------------------
# Encryption / Decryption
# ---------------------------------------------------------------------------
def aes_encrypt(plaintext, key_hex, iv_hex):try:key_bytes = binascii.unhexlify(key_hex)iv_bytes = binascii.unhexlify(iv_hex)b64_str = base64.b64encode(plaintext.encode('utf-8')).decode('utf-8')data_bytes = b64_str.encode('utf-8')padded_data = pad(data_bytes, AES.block_size)cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)ciphertext_bytes = cipher.encrypt(padded_data)return binascii.hexlify(ciphertext_bytes).decode('utf-8')except Exception as e:print(f"Encryption error: {e}")return Nonedef aes_decrypt_base64(ciphertext_hex, key_hex, iv_hex):if not ciphertext_hex:return ""try:key_bytes = binascii.unhexlify(key_hex)iv_bytes = binascii.unhexlify(iv_hex)ct_bytes = binascii.unhexlify(ciphertext_hex)cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)decrypted = cipher.decrypt(ct_bytes)decrypted = unpad(decrypted, AES.block_size)b64_str = decrypted.decode('utf-8')plain_bytes = base64.b64decode(b64_str)return plain_bytes.decode('utf-8')except Exception:return ""def parse_time_arg(time_str):if not time_str:return Noneif 'T' in time_str:if '+' not in time_str and 'Z' not in time_str:return time_str + "+08:00"return time_strtry:dt = Nonetime_str = time_str.strip()if len(time_str) == 10:dt = datetime.strptime(time_str, "%Y-%m-%d")elif len(time_str) == 16:dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M")elif len(time_str) == 19:dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")if dt:return dt.strftime("%Y-%m-%dT%H:%M:%S+08:00")except ValueError:passreturn time_str# ---------------------------------------------------------------------------
# Data Fetching
# ---------------------------------------------------------------------------
def fetch_attendance_v3(start_time, end_time, aes_key, base_url, employee_no=None, proxies=None, fixed_iv=None):all_records = []# Determine IVif fixed_iv:iv = fixed_ivprint(f"Using Fixed IV: {iv}")else:# Dynamic IV based on timestamplogin_timestamp = int(time.time() * 1000)iv = hashlib.md5(str(login_timestamp).encode()).hexdigest()print(f"Using Dynamic IV: {iv}")url = f"{base_url}/ISAPI/AccessControl/AcsEvent?format=json&security=1&iv={iv}"position = 0batch_size = 24search_id = str(uuid.uuid4())print(f"Fetching data from {start_time} to {end_time}...")target_encrypted_emp = Noneif employee_no:if len(employee_no) > 20 and all(c in '0123456789abcdefABCDEF' for c in employee_no):target_encrypted_emp = employee_noelse:target_encrypted_emp = aes_encrypt(employee_no, aes_key, iv)print(f" -> Encrypted Employee No: {target_encrypted_emp}")while True:acs_event_cond = {"searchID": search_id,"searchResultPosition": position,"maxResults": batch_size,"major": 0,"minor": 0,"startTime": start_time,"endTime": end_time,}if target_encrypted_emp:acs_event_cond["employeeNoString"] = target_encrypted_emppayload = {"AcsEventCond": acs_event_cond}headers = {"Content-Type": "application/json","X-Requested-With": "XMLHttpRequest",}try:resp = requests.post(url, json=payload, headers=headers, auth=DIGEST_AUTH, proxies=proxies,timeout=30)resp.raise_for_status()data = resp.json()acs = data.get('AcsEvent', {})info = acs.get('InfoList', [])if not info:breakfor item in info:name_enc = item.get('name', '')emp_enc = item.get('employeeNoString', '')name = aes_decrypt_base64(name_enc, aes_key, iv) if len(name_enc) > 20 else name_enccurr_employee_no = aes_decrypt_base64(emp_enc, aes_key, iv) if len(emp_enc) > 20 else emp_encrecord = {'employeeNo': curr_employee_no,'name': name,'time': item.get('time', ''),'cardNo': item.get('cardNo', ''),'raw_employeeNoString': emp_enc }all_records.append(record)position += len(info)total_matches = acs.get('totalMatches', 0)print(f"Fetched {position}/{total_matches} records...")if position >= total_matches:breakexcept Exception as e:print(json.dumps({"error": f"Fetch error: {e}"}, ensure_ascii=False))breakreturn all_records# ---------------------------------------------------------------------------
# Main Function
# ---------------------------------------------------------------------------
def main_v3():global DIGEST_AUTHparser = argparse.ArgumentParser(description="Hikvision Attendance Fetcher V3 (Dynamic Key)")parser.add_argument("--start", help="Start time (e.g. 2025-12-08 08:00)", default=None)parser.add_argument("--end", help="End time (e.g. 2025-12-08 23:59)", default=None)parser.add_argument("--employeeNo", help="Employee Number (e.g. 16128 or encrypted hex)", default=None)parser.add_argument("--out", help="Output JSON file", default="attendance_v3.json")parser.add_argument("--ip", help=f"Target IP (default: {DEFAULT_IP})", default=DEFAULT_IP)parser.add_argument("--port", help=f"Target Port (default: {DEFAULT_PORT})", default=DEFAULT_PORT)parser.add_argument("--proxy", help="Proxy URL (e.g. http://127.0.0.1:8899)", default=None)parser.add_argument("--iv", help="Fixed IV (optional, for encrypted employeeNo)", default=None)args = parser.parse_args()host_str = f"{args.ip}:{args.port}"base_url = f"http://{host_str}"# Configure Proxiesproxies = Noneif args.proxy:proxies = {"http": args.proxy,"https": args.proxy,}print(f"Using Proxy: {args.proxy}")DIGEST_AUTH = HTTPDigestAuth(USERNAME, PASSWORD)# 1. Derive Keyprint("Deriving AES Key...")aes_key = get_aes_key(host_str, USERNAME, PASSWORD, DIGEST_AUTH, proxies)if not aes_key:print("Failed to derive key. Exiting.")return# 2. Parse Timesif args.start:args.start = parse_time_arg(args.start)else:now = datetime.now()start_dt = datetime(now.year, now.month, now.day, 0, 0, 0)args.start = start_dt.strftime("%Y-%m-%dT%H:%M:%S+08:00")if args.end:args.end = parse_time_arg(args.end)else:now = datetime.now()end_dt = datetime(now.year, now.month, now.day, 23, 59, 59)args.end = end_dt.strftime("%Y-%m-%dT%H:%M:%S+08:00")# 3. Fetch Datarecords = fetch_attendance_v3(args.start, args.end, aes_key, base_url, args.employeeNo, proxies, args.iv)# 4. Savewith open(args.out, 'w', encoding='utf-8') as f:json.dump(records, f, ensure_ascii=False, indent=2)print(f"Data successfully saved to {args.out} ({len(records)} records)")if __name__ == "__main__":main_v3()
)
