Spaces:
Paused
Paused
File size: 1,866 Bytes
5a3b9db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | """
Implements: 03_AI/00_AI_ARCHITECTURE.md (Rule Engine)
"""
import pandas as pd
import logging
from typing import Dict, Any
logger = logging.getLogger(__name__)
class RuleEngine:
"""
Deterministic rule-based security heuristics evaluator.
"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.velocity_threshold = config.get("velocity_threshold", 5) # seconds
self.failure_threshold = config.get("failure_threshold", 5)
def evaluate(self, features: pd.DataFrame) -> pd.DataFrame:
logger.info("Evaluating deterministic rules...")
# Evaluate Rule: Excessive Failed Logins
excessive_failures = features["rolling_failures_24h"] >= self.failure_threshold
# Evaluate Rule: Login Velocity Threshold
velocity_anomaly = features["time_since_last_login"] < self.velocity_threshold
# Evaluate Rule: Off-hours Authentication
off_hours = features["is_working_hour"] == 0
# Generate base rule score
rule_score = (excessive_failures.astype(int) * 30 +
velocity_anomaly.astype(int) * 20 +
off_hours.astype(int) * 10)
rule_score = rule_score.clip(upper=100)
triggered_rules = []
for i in range(len(features)):
rules = []
if excessive_failures.iloc[i]: rules.append("ExcessiveFailures")
if velocity_anomaly.iloc[i]: rules.append("VelocityAnomaly")
if off_hours.iloc[i]: rules.append("OffHours")
triggered_rules.append(",".join(rules) if rules else "None")
return pd.DataFrame({
"event_id": features["event_id"],
"rule_score": rule_score,
"triggered_rules": triggered_rules
})
|