#!/usr/bin/env python3 import requests import json import datetime import sys import os from urllib3.exceptions import InsecureRequestWarning # Suppress insecure request warnings (Self-signed certs on firewall) requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) # ================= CONFIGURATION ================= # Toggle Debug Messages (True = Show detailed API responses, False = Clean output) DEBUG_MODE = False # Firewall Details NSF_HOST = "192.168.1.1" # IP ของอุปกรณ์ NSF (เช่น 10.1.1.254) NSF_PORT = 4433 # พอร์ต WebUI/Management (ค่าปกติคือ 443 หรือ 4433) NSF_USER = "admin" # ชื่อผู้ใช้ระดับ Admin (แนะนำให้สร้าง user แยกสำหรับการทำ API) NSF_PASS = "your_password" # รหัสผ่าน # Certificate Files (Standard Let's Encrypt paths) # ระบุ path ของไฟล์ certificate ที่ได้จาก Let's Encrypt CERT_PATH = "/etc/letsencrypt/live/yourdomain.com/fullchain.pem" KEY_PATH = "/etc/letsencrypt/live/yourdomain.com/privkey.pem" # The Exact Name of the Decryption Policy to Update # ชื่อของ Decryption Policy ที่ต้องการให้อัปเดต (ต้องตรงกับในหน้า Web UI เป๊ะๆ) TARGET_POLICY_NAME = "Your_Decryption_Policy_Name" # API Base URL BASE_URL = f"https://{NSF_HOST}:{NSF_PORT}/api/v1/namespaces/public" # ================================================= def debug_print(msg): """Prints message only if DEBUG_MODE is True.""" if DEBUG_MODE: print(msg) def read_file(filepath): """Reads the content of a file.""" try: with open(filepath, 'r') as f: return f.read() except FileNotFoundError: print(f"[!] Error: File not found at {filepath}") sys.exit(1) except Exception as e: print(f"[!] Error reading file: {str(e)}") sys.exit(1) def login(): """Logs in and returns the session token.""" url = f"{BASE_URL}/login" payload = { "name": NSF_USER, "password": NSF_PASS } print(f"[*] Authenticating with NSF at {NSF_HOST}:{NSF_PORT}...") try: response = requests.post(url, json=payload, verify=False, timeout=10) debug_print(f"[DEBUG] Login Status: {response.status_code}") # debug_print(f"[DEBUG] Login Response: {response.text}") response.raise_for_status() data = response.json() # รองรับโครงสร้าง Response หลายรูปแบบ if 'token' in data: return data['token'] elif 'loginResult' in data and 'token' in data['loginResult']: return data['loginResult']['token'] elif 'data' in data and 'loginResult' in data['data']: return data['data']['loginResult']['token'] else: print(f"[!] Login failed. Unexpected response structure: {data}") sys.exit(1) except Exception as e: print(f"[!] Login Error: {str(e)}") if 'response' in locals(): debug_print(f"[DEBUG] Failed Response Body: {response.text}") sys.exit(1) def verify_import(token, cert_name): """Checks if the cert exists in the list.""" url = f"{BASE_URL}/decryption/certs" headers = { "Content-Type": "application/json", "Cookie": f"token={token}" } print(f"[*] Verifying existence of '{cert_name}'...") try: response = requests.get(url, headers=headers, verify=False, timeout=10) debug_print(f"[DEBUG] List Certs Status: {response.status_code}") if response.status_code == 200: if cert_name in response.text: print(f"[+] Verification Successful: '{cert_name}' found on server.") return True else: print(f"[-] Verification Warning: '{cert_name}' NOT found in certificate list.") return False else: print(f"[!] Failed to list certificates. Code: {response.status_code}") debug_print(f"[DEBUG] Response: {response.text}") return False except Exception as e: print(f"[!] Verification Error: {str(e)}") return False def import_certificate(token, cert_content, key_content): """Imports the certificate and private key.""" # Create a unique name for the cert based on date date_str = datetime.datetime.now().strftime("%Y-%m-%d_%H%M") cert_name = f"LetsEncrypt_{date_str}" url = f"{BASE_URL}/decryption/certs/importkey" headers = { "Content-Type": "application/json", "Cookie": f"token={token}" } payload = { "name": cert_name, "passwd": "sangfor123", # Password dummy required by API (>4 chars) to store key securely "privateKey": { "keyType": "text", "text": key_content }, "publicKey": { "keyType": "text", "text": cert_content } } print(f"[*] Importing certificate as '{cert_name}'...") try: response = requests.post(url, headers=headers, json=payload, verify=False, timeout=30) # DEBUG: Always print response details debug_print(f"[DEBUG] Import Status Code: {response.status_code}") debug_print(f"[DEBUG] Import Response Body: {response.text}") if response.status_code == 200: resp_json = response.json() # API returns code 0 for success if resp_json.get('code') == 0: print("[+] Certificate import successful (Code 0).") return cert_name elif "success" in str(resp_json).lower() and "failed" not in str(resp_json).lower(): # Fallback check if 'code' isn't standard print("[+] Certificate import appears successful (Text check).") return cert_name else: print(f"[!] Import Logical Failure: {resp_json.get('message', 'Unknown error')}") return None else: print(f"[!] Import failed. Code: {response.status_code}") return None except Exception as e: print(f"[!] Import Error: {str(e)}") return None def update_decryption_policy(token, policy_name, cert_name): """Updates the decryption policy to use the new cert.""" url = f"{BASE_URL}/decryption/policies/{policy_name}" headers = { "Content-Type": "application/json", "Cookie": f"token={token}" } payload = { "certs": [cert_name] } print(f"[*] Updating policy '{policy_name}' to use '{cert_name}'...") try: # Try PATCH first debug_print(f"[DEBUG] Sending PATCH to {url}") response = requests.patch(url, headers=headers, json=payload, verify=False, timeout=10) debug_print(f"[DEBUG] Update (PATCH) Status: {response.status_code}") debug_print(f"[DEBUG] Update (PATCH) Body: {response.text}") # Fallback to PUT if PATCH isn't allowed (Some firmware versions) if response.status_code == 405: print("[*] PATCH method not allowed, trying PUT...") response = requests.put(url, headers=headers, json=payload, verify=False, timeout=10) debug_print(f"[DEBUG] Update (PUT) Status: {response.status_code}") debug_print(f"[DEBUG] Update (PUT) Body: {response.text}") if response.status_code == 200 and response.json().get('code') == 0: print(f"[+] Policy '{policy_name}' updated successfully!") else: print(f"[!] Policy update failed: {response.text}") sys.exit(1) except Exception as e: print(f"[!] Policy Update Error: {str(e)}") sys.exit(1) def main(): if DEBUG_MODE: print("--- Starting NSF Certificate Update (DEBUG MODE) ---") else: print("--- Starting NSF Certificate Update ---") fullchain = read_file(CERT_PATH) privkey = read_file(KEY_PATH) token = login() new_cert_name = import_certificate(token, fullchain, privkey) if not new_cert_name: print("[!] Aborting: Certificate import failed.") sys.exit(1) exists = verify_import(token, new_cert_name) if not exists: print("[!] Warning: Certificate was not found in the list after import.") print("[!] Attempting to continue anyway, but policy update may fail...") update_decryption_policy(token, TARGET_POLICY_NAME, new_cert_name) print("--- Finished ---") if __name__ == "__main__": main()