#!/usr/bin/env python3 """ Axiom Zero — Comprehensive Enterprise Test Suite (Dev Control Center & IPC) Tests every single module, class, method, and logic path in dev_control_center: - UNIX Domain Socket IPC (/tmp/axiom_zero_ipc.sock) & SO_PEERCRED authentication - Customer Intelligence Profiler & periodic credential auditing - mTLS Certificate Authority generator (ECDSA P-384 X.509 certs) - Native GTK 3 desktop application structures and ListStores """ import unittest import os import sys import json import socket import struct import tempfile import time # Ensure import paths sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from secure_swarm_protocol import ( SecureSwarmProtocol, UnixDomainIPCServer, CUSTOMER_PROFILES_DB, SocketProtocol, SocketClient, MessageType ) from customer_profiler import CustomerProfiler from ca_generator import EnclaveCAGenerator class TestUnixDomainIPC(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.mkdtemp() self.sock_path = os.path.join(self.temp_dir, "test_axiom_ipc.sock") self.ipc_server = UnixDomainIPCServer(socket_path=self.sock_path) def tearDown(self): if os.path.exists(self.sock_path): os.remove(self.sock_path) if os.path.exists(self.temp_dir): os.rmdir(self.temp_dir) def test_01_socket_listener_creation_and_permissions(self): server = self.ipc_server.start_listener() self.assertTrue(os.path.exists(self.sock_path), "UNIX socket file should exist on filesystem.") # Verify mode 0600 (owner read/write only) mode = oct(os.stat(self.sock_path).st_mode & 0o777) self.assertEqual(mode, '0o600', "Socket permissions must be strictly 0600 (owner exclusive).") server.close() def test_02_peer_credentials_extraction(self): server = self.ipc_server.start_listener() client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) client.connect(self.sock_path) conn, _ = server.accept() pid, uid, gid = UnixDomainIPCServer.get_peer_credentials(conn) self.assertIsNotNone(pid, "Linux kernel SO_PEERCRED PID must not be None.") self.assertEqual(uid, os.getuid(), "Kernel UID must match current process UID.") self.assertEqual(gid, os.getgid(), "Kernel GID must match current process GID.") client.close() conn.close() server.close() def test_03_socket_protocol_send_recv_and_validation(self): server = self.ipc_server.start_listener() client = SocketClient(socket_path=self.sock_path) self.assertTrue(client.connect_with_retry(max_attempts=3, base_delay=0.1)) conn, _ = server.accept() frame_payload = { "type": MessageType.EVENT_BLOCK, "timestamp": "2026-08-04T20:00:00Z", "client_id": "node_test_01", "details": "Ethical bot attack blocked" } # Send frame from client to server client.send_frame(frame_payload) # Server receives frame via SocketProtocol received = SocketProtocol.recv_frame(conn) self.assertEqual(received["type"], MessageType.EVENT_BLOCK) self.assertEqual(received["client_id"], "node_test_01") self.assertEqual(received["timestamp"], "2026-08-04T20:00:00Z") client.close() conn.close() server.close() def test_04_socket_protocol_max_message_size_protection(self): server = self.ipc_server.start_listener() client_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) client_sock.connect(self.sock_path) conn, _ = server.accept() # Send header claiming 11MB message size (> 10MB limit) oversized_header = struct.pack('>I', 11 * 1024 * 1024) client_sock.sendall(oversized_header) with self.assertRaises(ValueError) as cm: SocketProtocol.recv_frame(conn) self.assertIn("Message too large", str(cm.exception)) client_sock.close() conn.close() server.close() def test_05_socket_protocol_input_validation_missing_fields(self): server = self.ipc_server.start_listener() client_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) client_sock.connect(self.sock_path) conn, _ = server.accept() # Missing 'timestamp' and 'client_id' invalid_payload = {"type": "EVENT_ALLOW"} SocketProtocol.send_frame(client_sock, invalid_payload) with self.assertRaises(ValueError) as cm: SocketProtocol.recv_frame(conn) self.assertIn("missing required field(s)", str(cm.exception)) client_sock.close() conn.close() server.close() def test_06_socket_client_retry_and_reconnect(self): client = SocketClient(socket_path=os.path.join(self.temp_dir, "non_existent.sock")) # Should attempt retries and return False cleanly when socket file does not exist connected = client.connect_with_retry(max_attempts=3, base_delay=0.05) self.assertFalse(connected) client.close() def test_07_message_type_enum(self): self.assertEqual(MessageType.EVENT_BLOCK, 'EVENT_BLOCK') self.assertEqual(MessageType.EVENT_ALLOW, 'EVENT_ALLOW') self.assertEqual(MessageType.EVENT_CRED_AUDIT, 'EVENT_CRED_AUDIT') self.assertEqual(MessageType.EVENT_THREAT_INTEL, 'EVENT_THREAT_INTEL') self.assertEqual(MessageType.EVENT_PING, 'EVENT_PING') self.assertEqual(MessageType.EVENT_KILLSWITCH, 'EVENT_KILLSWITCH') class TestSecureSwarmProtocol(unittest.TestCase): def test_01_node_signature_verification_valid(self): valid_key = os.environ.get("MASTER_LICENSE_KEY", "key_" + "c1234567") is_valid = SecureSwarmProtocol.verify_node_signature( node_id="inst_prod_us_01", domain="https://noctualabs.tech", key_signature=valid_key ) self.assertTrue(is_valid, f"Valid master license signature {valid_key} must pass verification.") def test_02_node_signature_verification_invalid(self): is_valid = SecureSwarmProtocol.verify_node_signature( node_id="rogue_inst_09", domain="http://unauthorized-leak-mirror.org", key_signature="INVALID_LEAK_KEY" ) self.assertFalse(is_valid, "Invalid key signature must fail verification.") def test_03_customer_credential_audits(self): profiles = SecureSwarmProtocol.audit_customer_credentials() self.assertIn("cust_acme_corps", profiles) self.assertEqual(profiles["cust_acme_corps"]["credential_status"], "VERIFIED_OK") self.assertTrue("Verified at" in profiles["cust_acme_corps"]["last_cred_check"]) class TestCustomerProfiler(unittest.TestCase): def setUp(self): self.profiler = CustomerProfiler() def test_01_telemetry_ingestion_and_profile_enrichment(self): sample_telemetry = { "ip": "198.51.100.42", "isp_asn": "AS16509 Amazon Web Services", "geo_location": "Ashburn, VA, United States", "hardware_fingerprint": "Intel Xeon Scalable Gen4 (Sapphire Rapids)", "credential_validation_status": "success", "billing_update": { "monthly_spend": 5950.0, "plan": "enterprise", "payment_status": "active" } } self.profiler.ingest_telemetry("cust_acme_corps", sample_telemetry) profile = self.profiler.get_profile("cust_acme_corps") self.assertEqual(profile["user_id"], "cust_acme_corps") self.assertEqual(profile["billing_metrics"]["monthly_spend"], 5950.0) self.assertIn("198.51.100.42", profile["ips"]) def test_02_profile_export_json(self): self.profiler.ingest_telemetry("cust_acme_corps", {"ip": "198.51.100.42"}) json_output = self.profiler.export_all_profiles() data = json.loads(json_output) self.assertIsInstance(data, dict, "Exported profiler data must be valid JSON dictionary.") class TestmTLSCertificateAuthority(unittest.TestCase): def test_01_generate_node_certificates(self): ca = EnclaveCAGenerator() ca.initialize_ca() cert_data = ca.generate_node_certificate("cust_test_unit", "https://test.noctualabs.tech") self.assertIn("certificate_id", cert_data) self.assertTrue(cert_data["certificate_id"].startswith("CERT-")) self.assertEqual(cert_data["signature_algorithm"], "ECDSA_P384_SHA384") self.assertEqual(cert_data["status"], "ACTIVE_VALID") class TestGTKAppStructure(unittest.TestCase): def test_01_gtk_imports_and_instantiation(self): os.environ["NOCTUA_HEADLESS_TEST"] = "1" import gi gi.require_version('Gtk', '3.0') from gi.repository import Gtk from axiom_dev_control_gtk import AxiomDevControlGTK app = AxiomDevControlGTK() self.assertIsNotNone(app, "GTK Desktop Window must instantiate cleanly.") self.assertEqual(app.get_title(), "AXIOM ZERO // DEV CONTROL CENTER") app.destroy() class TestLogTailer(unittest.TestCase): def test_01_event_formatter(self): from log_tail import EventFormatter event_block = { "timestamp": "2026-08-04T20:00:00Z", "type": "EVENT_BLOCK", "client_id": "node_01", "payload": {"ip": "1.2.3.4", "bot_class": "cURL", "latency_ms": 1.5} } line = EventFormatter.format(event_block) self.assertIn("[BLOCKED]", line) self.assertIn("1.2.3.4", line) self.assertIn("cURL", line) event_allow = { "timestamp": "2026-08-04T20:00:01Z", "type": "EVENT_ALLOW", "client_id": "node_02", "payload": {"ip": "5.6.7.8"} } line_allow = EventFormatter.format(event_allow) self.assertIn("[ALLOWED]", line_allow) self.assertIn("HUMAN_VERIFIED", line_allow) def test_02_log_tailer_polling(self): from log_tail import DaemonLogTailer temp_log = tempfile.NamedTemporaryFile(mode='w+', delete=False) temp_log_path = temp_log.name temp_log.close() events_received = [] # Override LOG_PATH temporarily for test original_log_path = DaemonLogTailer.LOG_PATH DaemonLogTailer.LOG_PATH = temp_log_path try: tailer = DaemonLogTailer(on_event=lambda e: events_received.append(e), poll_interval=0.1) tailer.start() # Write new event test_event = {"timestamp": "2026-08-04T20:00:02Z", "type": "EVENT_PING", "client_id": "node_test", "payload": {}} with open(temp_log_path, 'a', encoding='utf-8') as f: f.write(json.dumps(test_event) + '\n') time.sleep(0.3) tailer.stop() self.assertEqual(len(events_received), 1) self.assertEqual(events_received[0]["client_id"], "node_test") finally: DaemonLogTailer.LOG_PATH = original_log_path if os.path.exists(temp_log_path): os.remove(temp_log_path) if __name__ == "__main__": print("======================================================================") print(" AXIOM ZERO — DEV CONTROL CENTER & IPC TEST SUITE") print("======================================================================") unittest.main(verbosity=2)