import os import math import hashlib import numpy as np from scipy import stats import joblib import warnings try: from scipy.special import entr except ImportError: pass try: from sklearn.ensemble import IsolationForest except ImportError: print("Error: sklearn is not installed. Please install scikit-learn.") import sys sys.exit(1) # Suppress warnings from constant arrays in stats warnings.filterwarnings('ignore', category=stats.ConstantInputWarning) warnings.filterwarnings('ignore', category=RuntimeWarning) def _compute_sha256(filepath: str) -> str: h = hashlib.sha256() with open(filepath, 'rb') as f: while chunk := f.read(65536): h.update(chunk) return h.hexdigest() class KinematicsFeatureExtractor: """Extracts features from mouse/touch trajectories.""" @staticmethod def extract(path: list) -> dict: """ Input: list of (x, y, timestamp_ms) tuples representing a mouse/touch path """ if len(path) < 2: return { 'velocity_mean': 0.0, 'velocity_std': 0.0, 'velocity_max': 0.0, 'angular_velocity_mean': 0.0, 'angular_velocity_std': 0.0, 'bezier_entropy': 0.0, 'inter_event_jitter_cv': 0.0, 'fitts_compliance': 0.0, 'acceleration_sign_changes': 0, 'pause_count': 0, 'total_path_length': 0.0, 'path_efficiency': 0.0, 'tsallis_entropy_q2': 0.0 } velocities = [] angular_velocities = [] angles = [] accelerations = [] inter_event_times = [] pauses = 0 total_len = 0.0 for i in range(1, len(path)): x1, y1, t1 = path[i-1] x2, y2, t2 = path[i] dt = max(t2 - t1, 1) # avoid div by zero dx = x2 - x1 dy = y2 - y1 dist = math.hypot(dx, dy) total_len += dist v = dist / dt velocities.append(v) inter_event_times.append(dt) if dt > 200: pauses += 1 angle = math.degrees(math.atan2(dy, dx)) angles.append(angle) for i in range(1, len(velocities)): dv = velocities[i] - velocities[i-1] dt = max(path[i+1][2] - path[i][2], 1) accelerations.append(dv / dt) d_angle = abs(angles[i] - angles[i-1]) if d_angle > 180: d_angle = 360 - d_angle angular_velocities.append(d_angle / dt) v_mean = np.mean(velocities) if velocities else 0.0 v_std = np.std(velocities) if velocities else 0.0 v_max = np.max(velocities) if velocities else 0.0 av_mean = np.mean(angular_velocities) if angular_velocities else 0.0 av_std = np.std(angular_velocities) if angular_velocities else 0.0 acc_sign_changes = 0 for i in range(1, len(accelerations)): if (accelerations[i] > 0 > accelerations[i-1]) or (accelerations[i] < 0 < accelerations[i-1]): acc_sign_changes += 1 cv = (np.std(inter_event_times) / np.mean(inter_event_times)) if inter_event_times and np.mean(inter_event_times) > 0 else 0.0 start_end_dist = math.hypot(path[-1][0] - path[0][0], path[-1][1] - path[0][1]) eff = (start_end_dist / total_len) if total_len > 0 else 1.0 fitts_corr = 0.0 if len(path) >= 3: distances = [math.hypot(path[i][0]-path[i-1][0], path[i][1]-path[i-1][1]) for i in range(1, len(path))] times = [max(path[i][2]-path[i-1][2], 1) for i in range(1, len(path))] ids = [math.log2(d + 1) for d in distances] if len(ids) >= 2 and np.std(ids) > 1e-6 and np.std(times) > 1e-6: corr, _ = stats.pearsonr(ids, times) if not np.isnan(corr): fitts_corr = corr # Approximate bezier entropy using positional variance as proxy bezier_entropy = np.var(angles) if len(angles) > 0 else 0.0 # Tsallis entropy q=2: (1 - sum(p_i^2)) if velocities: counts, _ = np.histogram(velocities, bins=10) total_counts = np.sum(counts) if total_counts > 0: p = counts / total_counts p = p[p > 0] tsallis = 1.0 - np.sum(p**2) else: tsallis = 0.0 else: tsallis = 0.0 return { 'velocity_mean': float(v_mean), 'velocity_std': float(v_std), 'velocity_max': float(v_max), 'angular_velocity_mean': float(av_mean), 'angular_velocity_std': float(av_std), 'bezier_entropy': float(bezier_entropy), 'inter_event_jitter_cv': float(cv), 'fitts_compliance': float(fitts_corr), 'acceleration_sign_changes': int(acc_sign_changes), 'pause_count': int(pauses), 'total_path_length': float(total_len), 'path_efficiency': float(eff), 'tsallis_entropy_q2': float(tsallis) } class KeystrokeCadenceScorer: @staticmethod def extract(keystrokes: list) -> dict: """ Input: list of (key, press_time_ms, release_time_ms) tuples """ if len(keystrokes) < 2: return { 'dwell_time_mean': 0.0, 'dwell_time_std': 0.0, 'flight_time_mean': 0.0, 'flight_time_std': 0.0, 'digraph_consistency': 0.0, 'rhythm_autocorrelation': 0.0 } dwell_times = [] flight_times = [] inter_press = [] for i in range(len(keystrokes)): k1, p1, r1 = keystrokes[i] dwell_times.append(r1 - p1) if i < len(keystrokes) - 1: k2, p2, r2 = keystrokes[i+1] flight_times.append(p2 - r1) inter_press.append(p2 - p1) dwell_mean = np.mean(dwell_times) dwell_std = np.std(dwell_times) flight_mean = np.mean(flight_times) if flight_times else 0.0 flight_std = np.std(flight_times) if flight_times else 0.0 digraph_consistency = flight_std / flight_mean if flight_mean > 0 else 0.0 autocorr = 0.0 if len(inter_press) > 2: if np.std(inter_press[:-1]) > 1e-6 and np.std(inter_press[1:]) > 1e-6: c = np.corrcoef(inter_press[:-1], inter_press[1:])[0, 1] if not np.isnan(c): autocorr = c return { 'dwell_time_mean': float(dwell_mean), 'dwell_time_std': float(dwell_std), 'flight_time_mean': float(flight_mean), 'flight_time_std': float(flight_std), 'digraph_consistency': float(digraph_consistency), 'rhythm_autocorrelation': float(autocorr) } class BehaviorAnomalyDetector: def __init__(self): self.model = IsolationForest(contamination=0.5, random_state=42, n_estimators=100) self.feature_names = [ 'velocity_mean', 'velocity_std', 'velocity_max', 'angular_velocity_mean', 'angular_velocity_std', 'bezier_entropy', 'inter_event_jitter_cv', 'fitts_compliance', 'acceleration_sign_changes', 'pause_count', 'total_path_length', 'path_efficiency', 'tsallis_entropy_q2', 'dwell_time_mean', 'dwell_time_std', 'flight_time_mean', 'flight_time_std', 'digraph_consistency', 'rhythm_autocorrelation' ] def _dict_to_array(self, features: dict) -> np.ndarray: arr = np.array([[features.get(k, 0.0) for k in self.feature_names]], dtype=float) return np.nan_to_num(arr, nan=0.0, posinf=1.0, neginf=-1.0) def train(self, feature_matrix: np.ndarray): feature_matrix = np.nan_to_num(feature_matrix, nan=0.0, posinf=1.0, neginf=-1.0) self.model.fit(feature_matrix) def score(self, feature_vector: np.ndarray) -> float: # returns anomaly score 0.0 (human) to 1.0 (bot) feature_vector = np.nan_to_num(feature_vector, nan=0.0, posinf=1.0, neginf=-1.0) # IsolationForest decision_function returns negative for anomalies, positive for normal raw_score = self.model.decision_function(feature_vector)[0] # Normalize to 0-1, where 1 is highest bot probability # typically decision_function is roughly between -0.5 and 0.5 norm_score = 1.0 - (1.0 / (1.0 + np.exp(-10 * raw_score))) # Sigmoid flip return float(norm_score) def save_model(self, path: str): joblib.dump(self.model, path) sha256_hash = _compute_sha256(path) with open(f"{path}.sha256", 'w') as f: f.write(sha256_hash) def load_model(self, path: str): hash_path = f"{path}.sha256" if not os.path.exists(hash_path): raise ValueError(f"Security error: SHA256 hash file missing for model at {path}") with open(hash_path, 'r') as f: expected_hash = f.read().strip() actual_hash = _compute_sha256(path) if actual_hash != expected_hash: raise ValueError(f"Security error: SHA256 hash mismatch for model file {path}") self.model = joblib.load(path) def auto_train(self, model_path: str): hash_path = f"{model_path}.sha256" if os.path.exists(model_path) and os.path.exists(hash_path): try: self.load_model(model_path) return except Exception as e: print(f"Warning: Failed to load existing model ({e}). Re-training...") print("No valid model found. Generating synthetic training data...") np.random.seed(42) X_train = [] # Human synthetic (500) for _ in range(500): h = { 'velocity_mean': np.random.normal(0.5, 0.1), 'velocity_std': np.random.normal(0.2, 0.05), 'velocity_max': np.random.normal(2.0, 0.5), 'angular_velocity_mean': np.random.normal(45, 10), 'angular_velocity_std': np.random.normal(30, 5), 'bezier_entropy': np.random.normal(500, 100), 'inter_event_jitter_cv': np.random.normal(0.8, 0.2), 'fitts_compliance': np.random.normal(0.85, 0.1), 'acceleration_sign_changes': np.random.randint(10, 50), 'pause_count': np.random.randint(2, 10), 'total_path_length': np.random.normal(1500, 300), 'path_efficiency': np.random.normal(0.6, 0.1), 'tsallis_entropy_q2': np.random.normal(0.7, 0.1), 'dwell_time_mean': np.random.normal(100, 20), 'dwell_time_std': np.random.normal(30, 10), 'flight_time_mean': np.random.normal(150, 40), 'flight_time_std': np.random.normal(50, 15), 'digraph_consistency': np.random.normal(0.5, 0.1), 'rhythm_autocorrelation': np.random.normal(0.4, 0.15) } X_train.append([h[k] for k in self.feature_names]) # Bot synthetic (500) for _ in range(500): b = { 'velocity_mean': np.random.normal(1.2, 0.05), 'velocity_std': np.random.normal(0.01, 0.005), 'velocity_max': np.random.normal(1.25, 0.05), 'angular_velocity_mean': np.random.normal(5, 2), 'angular_velocity_std': np.random.normal(2, 1), 'bezier_entropy': np.random.normal(10, 5), 'inter_event_jitter_cv': np.random.normal(0.05, 0.02), 'fitts_compliance': np.random.normal(0.1, 0.1), 'acceleration_sign_changes': np.random.randint(0, 5), 'pause_count': np.random.randint(0, 2), 'total_path_length': np.random.normal(800, 100), 'path_efficiency': np.random.normal(0.95, 0.02), 'tsallis_entropy_q2': np.random.normal(0.1, 0.05), 'dwell_time_mean': np.random.normal(50, 2), 'dwell_time_std': np.random.normal(1, 0.5), 'flight_time_mean': np.random.normal(50, 2), 'flight_time_std': np.random.normal(1, 0.5), 'digraph_consistency': np.random.normal(0.02, 0.01), 'rhythm_autocorrelation': np.random.normal(0.95, 0.02) } X_train.append([b[k] for k in self.feature_names]) X_train = np.array(X_train) self.train(X_train) os.makedirs(os.path.dirname(model_path), exist_ok=True) self.save_model(model_path) print(f"Model trained and saved to {model_path}") MODEL_PATH = '/home/snuffleupagus/teamwork_projects/axiom_zero_audit/dev_control_center/behavioral_model.pkl' _detector_instance = None def get_or_train_model(model_path: str = MODEL_PATH) -> BehaviorAnomalyDetector: global _detector_instance if _detector_instance is None: _detector_instance = BehaviorAnomalyDetector() _detector_instance.auto_train(model_path) return _detector_instance def score_session(mouse_path: list, keystrokes: list) -> dict: """ Returns: { 'bot_probability': float, # 0.0-1.0 'risk_level': 'NONE'|'LOW'|'MEDIUM'|'HIGH'|'CRITICAL', 'features': dict, # all extracted features 'kinematics_score': float, 'keystroke_score': float, 'combined_score': float, 'decision': 'HUMAN'|'BOT' } """ detector = get_or_train_model(MODEL_PATH) kin_feats = KinematicsFeatureExtractor.extract(mouse_path) key_feats = KeystrokeCadenceScorer.extract(keystrokes) combined_feats = {**kin_feats, **key_feats} feat_vec = detector._dict_to_array(combined_feats) combined_score = detector.score(feat_vec) # Rough split for individual scores kin_vec = detector._dict_to_array({**kin_feats, **{k: 0 for k in key_feats.keys()}}) key_vec = detector._dict_to_array({**{k: 0 for k in kin_feats.keys()}, **key_feats}) kinematics_score = detector.score(kin_vec) keystroke_score = detector.score(key_vec) bot_probability = combined_score if bot_probability > 0.9: risk_level = 'CRITICAL' elif bot_probability > 0.7: risk_level = 'HIGH' elif bot_probability > 0.4: risk_level = 'MEDIUM' elif bot_probability > 0.2: risk_level = 'LOW' else: risk_level = 'NONE' decision = 'BOT' if bot_probability > 0.6 else 'HUMAN' return { 'bot_probability': bot_probability, 'risk_level': risk_level, 'features': combined_feats, 'kinematics_score': kinematics_score, 'keystroke_score': keystroke_score, 'combined_score': combined_score, 'decision': decision } if __name__ == '__main__': print("=== Axiom Zero Behavioral Kinematics Engine Demo ===") # Generate mock session data human_mouse = [(100+i*5+np.random.normal(0,2), 200+i*2+np.random.normal(0,2), i*15+np.random.normal(0,5)) for i in range(100)] human_keys = [('a', 1000+i*200, 1080+i*200+np.random.normal(0,10)) for i in range(10)] bot_mouse = [(100+i*10, 200+i*10, i*10) for i in range(100)] bot_keys = [('a', 1000+i*100, 1050+i*100) for i in range(10)] edge_mouse = [(100+i*8+np.random.normal(0,0.5), 200+i*5+np.random.normal(0,0.5), i*12) for i in range(100)] edge_keys = [('a', 1000+i*150, 1070+i*150) for i in range(10)] print("\n[1] Scoring Clear Human Session...") res1 = score_session(human_mouse, human_keys) print(f"Decision: {res1['decision']}, Probability: {res1['bot_probability']:.4f}, Risk: {res1['risk_level']}") print("\n[2] Scoring Clear Bot Session...") res2 = score_session(bot_mouse, bot_keys) print(f"Decision: {res2['decision']}, Probability: {res2['bot_probability']:.4f}, Risk: {res2['risk_level']}") print("\n[3] Scoring Edge Case Session...") res3 = score_session(edge_mouse, edge_keys) print(f"Decision: {res3['decision']}, Probability: {res3['bot_probability']:.4f}, Risk: {res3['risk_level']}")