danielostrow
/

c2sentinel

@@ -1,749 +0,0 @@
-# C2Sentinel API Reference
-Complete technical documentation for the C2Sentinel Python API.
-**Author:** Daniel Ostrow
-**Website:** [neuralintellect.com](https://neuralintellect.com)
----
-## Table of Contents
-1. [C2Sentinel Class](#c2sentinel-class)
-2. [AnalysisResult Class](#analysisresult-class)
-3. [ConnectionContext Class](#connectioncontext-class)
-4. [ReconSupport Class](#reconsupport-class)
-5. [FeatureExtractor Class](#featureextractor-class)
-6. [LogParser Class](#logparser-class)
-7. [Enums and Constants](#enums-and-constants)
----
-## C2Sentinel Class
-Main interface for C2 detection.
-### Constructor
-```python
-C2Sentinel(model: LogBERTC2Sentinel, config: C2SentinelConfig, device: str = 'auto')
-```
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `model` | LogBERTC2Sentinel | The neural network model |
-| `config` | C2SentinelConfig | Model configuration |
-| `device` | str | Device for inference ('auto', 'cpu', 'cuda') |
-### Class Methods
-#### load
-```python
-@classmethod
-def load(cls, path: str, device: str = 'auto') -> 'C2Sentinel'
-```
-Load a pre-trained model from safetensors format.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `path` | str | Path to model files (without extension) |
-| `device` | str | Device for inference |
-**Returns:** C2Sentinel instance
-**Example:**
-```python
-sentinel = C2Sentinel.load('c2_sentinel')
-sentinel = C2Sentinel.load('/path/to/c2_sentinel', device='cuda')
-```
-#### create_new
-```python
-@classmethod
-def create_new(cls, device: str = 'auto') -> 'C2Sentinel'
-```
-Create a new untrained model instance.
-**Returns:** C2Sentinel instance with random weights
----
-### Instance Methods
-#### analyze
-```python
-def analyze(
-    self,
-    connections: List[Dict],
-    threshold: float = 0.5,
-    context: Optional[ConnectionContext] = None,
-    include_features: bool = False,
-    strict_mode: bool = False
-) -> AnalysisResult
-```
-Analyze a list of connections for C2 activity.
-| Parameter | Type | Default | Description |
-|-----------|------|---------|-------------|
-| `connections` | List[Dict] | required | List of connection records |
-| `threshold` | float | 0.5 | Detection threshold (0.0-1.0) |
-| `context` | ConnectionContext | None | Optional context for enrichment |
-| `include_features` | bool | False | Include raw feature vector in result |
-| `strict_mode` | bool | False | Enforce minimum 0.7 threshold |
-**Returns:** AnalysisResult object
-**Connection Record Fields:**
-```python
-{
-    'timestamp': float,      # Required: Unix timestamp
-    'dst_ip': str,           # Required: Destination IP
-    'dst_port': int,         # Required: Destination port
-    'bytes_sent': int,       # Required: Bytes sent
-    'bytes_recv': int,       # Required: Bytes received
-    'src_ip': str,           # Optional: Source IP
-    'src_port': int,         # Optional: Source port
-    'protocol': str,         # Optional: 'tcp' or 'udp'
-    'duration': float        # Optional: Duration in seconds
-}
-```
-**Example:**
-```python
-connections = [
-    {'timestamp': 1000, 'dst_ip': '10.0.0.1', 'dst_port': 443,
-     'bytes_sent': 200, 'bytes_recv': 500},
-    {'timestamp': 1060, 'dst_ip': '10.0.0.1', 'dst_port': 443,
-     'bytes_sent': 200, 'bytes_recv': 500},
-]
-result = sentinel.analyze(connections)
-result = sentinel.analyze(connections, threshold=0.7, strict_mode=True)
-```
----
-#### analyze_batch
-```python
-def analyze_batch(
-    self,
-    connection_groups: List[List[Dict]],
-    threshold: float = 0.5,
-    contexts: Optional[List[ConnectionContext]] = None,
-    parallel: bool = True
-) -> List[AnalysisResult]
-```
-Analyze multiple connection groups.
-| Parameter | Type | Default | Description |
-|-----------|------|---------|-------------|
-| `connection_groups` | List[List[Dict]] | required | List of connection lists |
-| `threshold` | float | 0.5 | Detection threshold |
-| `contexts` | List[ConnectionContext] | None | Context for each group |
-| `parallel` | bool | True | Enable parallel processing |
-**Returns:** List of AnalysisResult objects
-**Example:**
-```python
-groups = [
-    [conn1, conn2, conn3],
-    [conn4, conn5, conn6],
-]
-results = sentinel.analyze_batch(groups)
-```
----
-#### analyze_logs
-```python
-def analyze_logs(
-    self,
-    log_lines: List[str],
-    group_by_dst: bool = True,
-    threshold: float = 0.5
-) -> List[Dict]
-```
-Parse and analyze raw log lines.
-| Parameter | Type | Default | Description |
-|-----------|------|---------|-------------|
-| `log_lines` | List[str] | required | Raw log lines |
-| `group_by_dst` | bool | True | Group connections by destination IP |
-| `threshold` | float | 0.5 | Detection threshold |
-**Returns:** List of result dictionaries, sorted by probability (descending)
-**Supported Formats:**
-- JSON logs with standard fields
-- Zeek/Bro conn.log (tab-separated)
-- Syslog with IP:port patterns
-**Example:**
-```python
-with open('conn.log') as f:
-    lines = f.readlines()
-results = sentinel.analyze_logs(lines, group_by_dst=True)
-for r in results:
-    print(f"{r['dst_ip']}: {r['c2_probability']}")
-```
----
-#### add_whitelist
-```python
-def add_whitelist(
-    self,
-    ips: List[str] = None,
-    domains: List[str] = None
-)
-```
-Add IPs or domains to the whitelist. Whitelisted destinations receive reduced C2 probability.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `ips` | List[str] | IP addresses to whitelist |
-| `domains` | List[str] | Domain names to whitelist |
-**Example:**
-```python
-sentinel.add_whitelist(
-    ips=['8.8.8.8', '1.1.1.1'],
-    domains=['google.com', 'github.com']
-)
-```
----
-#### add_blacklist
-```python
-def add_blacklist(
-    self,
-    ips: List[str] = None,
-    domains: List[str] = None
-)
-```
-Add IPs or domains to the blacklist. Blacklisted destinations receive increased C2 probability.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `ips` | List[str] | IP addresses to blacklist |
-| `domains` | List[str] | Domain names to blacklist |
----
-#### save
-```python
-def save(self, path: str)
-```
-Save model to safetensors format.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `path` | str | Output path (without extension) |
-Creates two files:
-- `{path}.safetensors` - Model weights
-- `{path}.json` - Configuration
----
-### Instance Attributes
-| Attribute | Type | Description |
-|-----------|------|-------------|
-| `model` | LogBERTC2Sentinel | The neural network |
-| `config` | C2SentinelConfig | Model configuration |
-| `device` | torch.device | Inference device |
-| `feature_extractor` | FeatureExtractor | Feature extraction module |
-| `log_parser` | LogParser | Log parsing module |
-| `context_engine` | ContextInference | Context inference module |
-| `recon` | ReconSupport | Reconnaissance module |
----
-## AnalysisResult Class
-Dataclass containing analysis results.
-### Attributes
-| Attribute | Type | Description |
-|-----------|------|-------------|
-| `is_c2` | bool | True if C2 detected |
-| `c2_probability` | float | Probability score (0.0-1.0) |
-| `anomaly_score` | float | Anomaly detection score |
-| `evasion_score` | float | Evasion technique detection score |
-| `confidence` | float | Model confidence in prediction |
-| `c2_type` | str | Detected C2 framework type |
-| `c2_type_confidence` | float | Confidence in C2 type classification |
-| `detection_method` | str | Detection method used |
-| `immediate_detection` | bool | True if signature-based detection |
-| `context_applied` | bool | True if context was applied |
-| `original_probability` | float | Probability before context adjustment |
-| `probability_modifier` | float | Context probability modifier |
-| `matched_legitimate_pattern` | str | Name of matched legitimate pattern |
-| `legitimate_confidence` | float | Confidence in legitimate pattern match |
-| `risk_factors` | List[str] | Factors supporting C2 classification |
-| `mitigating_factors` | List[str] | Factors against C2 classification |
-| `service_type` | str | Detected service type |
-| `recommendations` | List[str] | Suggested follow-up actions |
-| `features` | List[float] | Raw 40-dimensional feature vector |
-### Methods
-#### to_dict
-```python
-def to_dict(self) -> Dict[str, Any]
-```
-Convert result to dictionary.
-**Returns:** Dictionary representation of all attributes
----
-## ConnectionContext Class
-Dataclass for providing additional context to improve detection accuracy.
-### Constructor
-```python
-ConnectionContext(
-    # Process information
-    process_name: Optional[str] = None,
-    process_path: Optional[str] = None,
-    process_pid: Optional[int] = None,
-    parent_process: Optional[str] = None,
-    command_line: Optional[str] = None,
-    # Network metadata
-    dns_queries: Optional[List[str]] = None,
-    resolved_hostname: Optional[str] = None,
-    tls_sni: Optional[str] = None,
-    tls_ja3: Optional[str] = None,
-    tls_ja3s: Optional[str] = None,
-    certificate_issuer: Optional[str] = None,
-    certificate_subject: Optional[str] = None,
-    certificate_valid: Optional[bool] = None,
-    http_user_agent: Optional[str] = None,
-    http_host: Optional[str] = None,
-    # Reputation
-    ip_reputation: Optional[float] = None,
-    domain_reputation: Optional[float] = None,
-    known_good: Optional[bool] = None,
-    known_bad: Optional[bool] = None,
-    threat_intel_match: Optional[str] = None,
-    # Host context
-    source_hostname: Optional[str] = None,
-    source_user: Optional[str] = None,
-    source_is_server: Optional[bool] = None,
-    source_is_workstation: Optional[bool] = None,
-    # Additional
-    geo_country: Optional[str] = None,
-    geo_asn: Optional[str] = None,
-    tags: Optional[List[str]] = None
-)
-```
-### Attribute Details
-| Attribute | Type | Effect on Analysis |
-|-----------|------|-------------------|
-| `process_name` | str | Known processes reduce probability |
-| `known_good` | bool | True reduces probability by 90% |
-| `known_bad` | bool | True increases probability by 5x |
-| `ip_reputation` | float | Score > 0.8 reduces probability |
-| `threat_intel_match` | str | Match increases probability by 5x |
-| `tls_ja3` | str | Known C2 JA3 increases probability |
-| `certificate_valid` | bool | False increases probability |
-### Methods
-#### to_dict
-```python
-def to_dict(self) -> Dict[str, Any]
-```
-Convert to dictionary, excluding None values.
----
-## ReconSupport Class
-Reconnaissance and enrichment utilities.
-### Class Methods
-#### analyze_ip
-```python
-@classmethod
-def analyze_ip(cls, ip: str) -> Dict[str, Any]
-```
-Analyze an IP address.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `ip` | str | IP address to analyze |
-**Returns:**
-```python
-{
-    'ip': str,              # Original IP
-    'is_valid': bool,       # Valid IP format
-    'is_private': bool,     # RFC 1918 private range
-    'is_loopback': bool,    # Loopback address
-    'is_multicast': bool,   # Multicast address
-    'is_cdn': bool,         # Known CDN range
-    'cdn_provider': str,    # CDN name if applicable
-    'ip_version': int,      # 4 or 6
-    'reverse_dns': str,     # Reverse DNS lookup result
-    'numeric': int          # Numeric representation
-}
-```
-**Known CDN Ranges:**
-- Cloudflare
-- AWS
-- Google Cloud
-- Azure
-- Akamai
----
-#### analyze_connection_patterns
-```python
-@classmethod
-def analyze_connection_patterns(cls, connections: List[Dict]) -> Dict[str, Any]
-```
-Analyze connection patterns for threat hunting.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `connections` | List[Dict] | Connection records |
-**Returns:**
-```python
-{
-    'connection_count': int,
-    'unique_destinations': int,
-    'unique_ports': int,
-    'timing': {
-        'duration_seconds': float,
-        'mean_interval': float,
-        'interval_stddev': float,
-        'interval_cv': float       # Coefficient of variation
-    },
-    'volume': {
-        'total_sent': int,
-        'total_recv': int,
-        'mean_sent': float,
-        'mean_recv': float,
-        'sent_recv_ratio': float
-    },
-    'ports': {
-        port_number: count,        # Port distribution
-        ...
-    },
-    'destinations': {
-        ip: analyze_ip_result,     # Per-IP analysis
-        ...
-    },
-    'indicators': {
-        'single_destination': bool,
-        'consistent_timing': bool,
-        'consistent_sizes': bool,
-        'uses_common_port': bool,
-        'uses_high_port': bool,
-        'has_cdn_destination': bool,
-        'all_private_destinations': bool
-    }
-}
-```
----
-#### generate_iocs
-```python
-@classmethod
-def generate_iocs(
-    cls,
-    connections: List[Dict],
-    result: Dict
-) -> Dict[str, List[str]]
-```
-Generate Indicators of Compromise from detected C2.
-| Parameter | Type | Description |
-|-----------|------|-------------|
-| `connections` | List[Dict] | Connection records |
-| `result` | Dict | Analysis result dictionary |
-**Returns:**
-```python
-{
-    'ips': List[str],                  # Destination IPs
-    'ports': List[str],                # Destination ports
-    'timing_signatures': List[str],    # Beacon timing patterns
-    'behavioral_indicators': List[str] # Behavioral markers
-}
-```
-Only generates IOCs if `result['is_c2']` is True.
----
-## FeatureExtractor Class
-Extracts 40-dimensional feature vectors from connections.
-### Constants
-#### C2_TYPES
-List of detectable C2 framework types:
-```python
-[
-    'unknown', 'metasploit', 'cobalt_strike', 'sliver', 'havoc',
-    'mythic', 'poshc2', 'merlin', 'empire', 'covenant',
-    'brute_ratel', 'koadic', 'pupy', 'silenttrinity', 'faction',
-    'ibombshell', 'godoh', 'dnscat2', 'iodine', 'dns_generic',
-    'http_custom', 'https_custom', 'websocket', 'domain_fronting',
-    'cloud_fronting', 'cdn_abuse', 'apt_generic', 'apt28', 'apt29',
-    'apt41', 'lazarus', 'fin7', 'turla', 'winnti', 'custom'
-]
-```
-### Methods
-#### extract_features
-```python
-def extract_features(self, connections: List[Dict]) -> np.ndarray
-```
-Extract 40-dimensional feature vector.
-**Returns:** numpy array of shape (40,)
-**Feature Groups:**
-- Features 0-9: Timing (intervals, jitter, regularity, periodicity)
-- Features 10-17: Destinations (diversity, persistence, ports)
-- Features 18-27: Payload (sizes, ratios, consistency)
-- Features 28-35: Evasion (jitter patterns, bursts, session length)
-- Features 36-39: Advanced (night activity, fast beacon ratio, duration)
----
-#### check_metasploit_signature
-```python
-def check_metasploit_signature(
-    self,
-    connections: List[Dict]
-) -> Tuple[bool, float]
-```
-Check for Metasploit-specific signature patterns.
-**Returns:** (is_metasploit, confidence)
----
-#### check_ssh_keepalive
-```python
-def check_ssh_keepalive(
-    self,
-    connections: List[Dict]
-) -> Tuple[bool, float]
-```
-Check for SSH keepalive pattern.
-**Criteria:**
-- Port 22
-- Small packets (< 100 bytes)
-- Symmetric traffic (sent/recv ratio 0.5-2.0)
-- Consistent sizes (CV < 0.2)
-- Regular intervals matching common keepalive values
-**Returns:** (is_ssh_keepalive, confidence)
----
-## LogParser Class
-Parses various log formats into connection records.
-### Static Methods
-#### parse_json
-```python
-@staticmethod
-def parse_json(log_line: str) -> Optional[Dict]
-```
-Parse JSON formatted log line.
-**Recognized Fields:**
-- timestamp, @timestamp
-- src_ip, source_ip, src
-- dst_ip, dest_ip, dst
-- src_port, source_port
-- dst_port, dest_port
-- bytes_sent, bytes_out
-- bytes_recv, bytes_in
----
-#### parse_zeek_conn
-```python
-@staticmethod
-def parse_zeek_conn(log_line: str) -> Optional[Dict]
-```
-Parse Zeek/Bro conn.log format (tab-separated).
----
-#### parse_syslog
-```python
-@staticmethod
-def parse_syslog(log_line: str) -> Optional[Dict]
-```
-Parse common syslog/netflow patterns.
-**Recognized Patterns:**
-- `YYYY-MM-DD HH:MM:SS ... IP:port -> IP:port`
-- `src=IP ... dst=IP ... sport=port ... dport=port`
----
-## Enums and Constants
-### DetectionMethod
-```python
-class DetectionMethod(Enum):
-    SIGNATURE = "signature"    # Port + behavior signature match
-    BEHAVIORAL = "behavioral"  # Pure behavioral analysis
-    ML = "ml"                  # Machine learning inference
-    CONTEXT = "context"        # Context-adjusted detection
-    HEURISTIC = "heuristic"    # Rule-based detection
-    WHITELIST = "whitelist"    # Matched whitelist pattern
-```
-### ServiceType
-```python
-class ServiceType(Enum):
-    SSH = "ssh"
-    HTTP = "http"
-    HTTPS = "https"
-    DNS = "dns"
-    DATABASE = "database"
-    API = "api"
-    STREAMING = "streaming"
-    GAMING = "gaming"
-    VPN = "vpn"
-    MONITORING = "monitoring"
-    UNKNOWN = "unknown"
-```
-### C2_INDICATOR_PORTS
-High-confidence C2 signature ports:
-```python
-{4444, 4445, 5555, 31337, 40056}
-```
-### C2_COMMON_PORTS
-Ports commonly used by C2 (require behavioral analysis):
-```python
-{80, 443, 53, 8080, 8443, 8888}
-```
----
-## Convenience Functions
-### load_model
-```python
-def load_model(path: str, device: str = 'auto') -> C2Sentinel
-```
-Shorthand for `C2Sentinel.load()`.
-### create_model
-```python
-def create_model(device: str = 'auto') -> C2Sentinel
-```
-Shorthand for `C2Sentinel.create_new()`.
-### quick_analyze
-```python
-def quick_analyze(
-    connections: List[Dict],
-    model_path: str = 'c2_sentinel'
-) -> AnalysisResult
-```
-One-shot analysis without keeping model in memory.
----
-## Error Handling
-The API uses standard Python exceptions:
-| Exception | Cause |
-|-----------|-------|
-| `FileNotFoundError` | Model files not found |
-| `ValueError` | Invalid connection format |
-| `RuntimeError` | CUDA/device errors |
-All methods handle empty or malformed input gracefully, returning neutral results rather than raising exceptions.