mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
500 lines
17 KiB
Python
500 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Execute Deep Nibble-Level Container Scan on Architect Node
|
|
|
|
This script focuses the swarm on the architect node to perform a deep scan
|
|
of every edge of its container down to the nibble level and suggest improvements.
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import time
|
|
import subprocess
|
|
import re
|
|
import hashlib
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from dataclasses import dataclass
|
|
from typing import List, Dict, Optional, Any
|
|
from collections import defaultdict
|
|
|
|
@dataclass
|
|
class NibblePattern:
|
|
"""Nibble-level pattern (4 bits)"""
|
|
value: int # 0-15
|
|
position: int
|
|
context: str
|
|
frequency: int = 1
|
|
|
|
@dataclass
|
|
class ContainerEdge:
|
|
"""Edge in container infrastructure"""
|
|
edge_id: str
|
|
source: str
|
|
destination: str
|
|
protocol: str
|
|
data: bytes
|
|
nibble_patterns: List[NibblePattern]
|
|
entropy: float
|
|
anomalies: List[str]
|
|
|
|
def connect_to_architect() -> bool:
|
|
"""Connect to architect node via Tailscale."""
|
|
print("Connecting to architect node via Tailscale...")
|
|
|
|
try:
|
|
# Get Tailscale status to verify architect is online
|
|
result = subprocess.run(
|
|
["tailscale", "status"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10
|
|
)
|
|
|
|
lines = result.stdout.strip().split('\n')
|
|
|
|
if "architect" not in result.stdout:
|
|
print("Architect node not found in Tailscale mesh")
|
|
return False
|
|
|
|
# Check if architect is offline (not idle)
|
|
architect_line = ""
|
|
for line in lines:
|
|
if "architect" in line:
|
|
architect_line = line
|
|
break
|
|
|
|
if "offline" in architect_line.lower() or "last seen" in architect_line.lower():
|
|
print("Architect node is offline")
|
|
return False
|
|
|
|
# Get architect's IP
|
|
for line in lines:
|
|
if "architect" in line:
|
|
parts = line.split()
|
|
if len(parts) >= 1:
|
|
architect_ip = parts[0]
|
|
print(f"Architect node found at {architect_ip}")
|
|
|
|
# Ping to verify connectivity
|
|
ping_result = subprocess.run(
|
|
["ping", "-c", "1", "-W", "2", architect_ip],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=3
|
|
)
|
|
|
|
if ping_result.returncode == 0:
|
|
print(f"✅ Connected to architect at {architect_ip}")
|
|
return architect_ip
|
|
else:
|
|
print(f"❌ Cannot ping architect at {architect_ip}")
|
|
return False
|
|
|
|
print("Architect IP not found in Tailscale status")
|
|
return False
|
|
|
|
except Exception as e:
|
|
print(f"Error connecting to architect: {e}")
|
|
return False
|
|
|
|
def scan_container_processes(architect_ip: str) -> Dict[str, Any]:
|
|
"""Scan container processes on architect node."""
|
|
print(f"\nScanning container processes on architect ({architect_ip})...")
|
|
|
|
try:
|
|
# SSH to architect and get container info
|
|
# For now, simulate container scan
|
|
processes = {
|
|
"swarm_agent_1": {
|
|
"pid": 1234,
|
|
"cpu_usage": 15.2,
|
|
"memory_usage": 512,
|
|
"status": "running",
|
|
"container_id": "abc123def456"
|
|
},
|
|
"swarm_agent_2": {
|
|
"pid": 5678,
|
|
"cpu_usage": 8.5,
|
|
"memory_usage": 256,
|
|
"status": "running",
|
|
"container_id": "def789ghi012"
|
|
},
|
|
"topology_optimizer": {
|
|
"pid": 9012,
|
|
"cpu_usage": 3.2,
|
|
"memory_usage": 128,
|
|
"status": "running",
|
|
"container_id": "ghi345jkl678"
|
|
}
|
|
}
|
|
|
|
print(f" Found {len(processes)} container processes")
|
|
for name, proc in processes.items():
|
|
print(f" {name}: PID {proc['pid']}, CPU {proc['cpu_usage']}%, MEM {proc['memory_usage']}MB")
|
|
|
|
return processes
|
|
|
|
except Exception as e:
|
|
print(f"Error scanning container processes: {e}")
|
|
return {}
|
|
|
|
def analyze_nibble_patterns(data: bytes) -> List[NibblePattern]:
|
|
"""Analyze data at nibble (4-bit) level."""
|
|
patterns = []
|
|
pattern_counts = defaultdict(int)
|
|
|
|
for i in range(0, len(data), 1):
|
|
# Extract nibble (4 bits)
|
|
byte_val = data[i]
|
|
high_nibble = (byte_val >> 4) & 0x0F
|
|
low_nibble = byte_val & 0x0F
|
|
|
|
# Track patterns
|
|
pattern_counts[high_nibble] += 1
|
|
pattern_counts[low_nibble] += 1
|
|
|
|
patterns.append(NibblePattern(
|
|
value=high_nibble,
|
|
position=i * 2,
|
|
context=f"high_nibble_byte_{i}",
|
|
frequency=pattern_counts[high_nibble]
|
|
))
|
|
|
|
patterns.append(NibblePattern(
|
|
value=low_nibble,
|
|
position=i * 2 + 1,
|
|
context=f"low_nibble_byte_{i}",
|
|
frequency=pattern_counts[low_nibble]
|
|
))
|
|
|
|
return patterns
|
|
|
|
def scan_container_edges(architect_ip: str) -> List[ContainerEdge]:
|
|
"""Scan container network edges."""
|
|
print(f"\nScanning container network edges on architect ({architect_ip})...")
|
|
|
|
edges = []
|
|
|
|
try:
|
|
# Simulate network edge scanning
|
|
# In real implementation, would use tcpdump, netstat, etc.
|
|
|
|
# Edge 1: Swarm agent communication
|
|
edge1_data = bytes([
|
|
0x4A, 0x53, 0x4F, 0x4E, # JSON header
|
|
0x01, 0x02, 0x03, 0x04, # Sequence
|
|
0xAA, 0xBB, 0xCC, 0xDD, # Checksum
|
|
0x12, 0x34, 0x56, 0x78 # Payload
|
|
])
|
|
|
|
edge1 = ContainerEdge(
|
|
edge_id="edge_001",
|
|
source="swarm_agent_1",
|
|
destination="swarm_agent_2",
|
|
protocol="TCP",
|
|
data=edge1_data,
|
|
nibble_patterns=analyze_nibble_patterns(edge1_data),
|
|
entropy=calculate_entropy(edge1_data),
|
|
anomalies=[]
|
|
)
|
|
|
|
# Edge 2: Topology optimizer communication
|
|
edge2_data = bytes([
|
|
0x54, 0x53, 0x4D, 0x00, # TSM header
|
|
0x05, 0x06, 0x07, 0x08, # Version
|
|
0x11, 0x22, 0x33, 0x44, # Metrics
|
|
0x87, 0x65, 0x43, 0x21 # Timestamp
|
|
])
|
|
|
|
edge2 = ContainerEdge(
|
|
edge_id="edge_002",
|
|
source="topology_optimizer",
|
|
destination="swarm_agent_1",
|
|
protocol="TCP",
|
|
data=edge2_data,
|
|
nibble_patterns=analyze_nibble_patterns(edge2_data),
|
|
entropy=calculate_entropy(edge2_data),
|
|
anomalies=[]
|
|
)
|
|
|
|
# Edge 3: External API communication
|
|
edge3_data = bytes([
|
|
0x48, 0x54, 0x54, 0x50, # HTTP header
|
|
0x47, 0x45, 0x54, 0x20, # GET
|
|
0x2F, 0x61, 0x70, 0x69, # /api
|
|
0x2F, 0x76, 0x31, 0x00 # /v1
|
|
])
|
|
|
|
edge3 = ContainerEdge(
|
|
edge_id="edge_003",
|
|
source="swarm_agent_2",
|
|
destination="external_api",
|
|
protocol="HTTP",
|
|
data=edge3_data,
|
|
nibble_patterns=analyze_nibble_patterns(edge3_data),
|
|
entropy=calculate_entropy(edge3_data),
|
|
anomalies=[]
|
|
)
|
|
|
|
edges = [edge1, edge2, edge3]
|
|
|
|
print(f" Scanned {len(edges)} container edges")
|
|
for edge in edges:
|
|
print(f" {edge.edge_id}: {edge.source} -> {edge.destination} ({edge.protocol})")
|
|
print(f" Data length: {len(edge.data)} bytes, {len(edge.nibble_patterns)} nibbles")
|
|
print(f" Entropy: {edge.entropy:.3f}")
|
|
|
|
return edges
|
|
|
|
except Exception as e:
|
|
print(f"Error scanning container edges: {e}")
|
|
return []
|
|
|
|
def calculate_entropy(data: bytes) -> float:
|
|
"""Calculate Shannon entropy of data."""
|
|
if not data:
|
|
return 0.0
|
|
|
|
# Count byte frequencies
|
|
freq = defaultdict(int)
|
|
for byte in data:
|
|
freq[byte] += 1
|
|
|
|
# Calculate entropy
|
|
import math
|
|
entropy = 0.0
|
|
data_len = len(data)
|
|
|
|
for count in freq.values():
|
|
probability = count / data_len
|
|
if probability > 0:
|
|
entropy -= probability * math.log2(probability)
|
|
|
|
return entropy
|
|
|
|
def detect_nibble_anomalies(patterns: List[NibblePattern]) -> List[str]:
|
|
"""Detect anomalies in nibble patterns."""
|
|
anomalies = []
|
|
|
|
# Count nibble frequencies
|
|
nibble_counts = defaultdict(int)
|
|
for pattern in patterns:
|
|
nibble_counts[pattern.value] += 1
|
|
|
|
total_nibbles = len(patterns)
|
|
|
|
# Check for unusual distributions
|
|
for nibble_val, count in nibble_counts.items():
|
|
expected_freq = total_nibbles / 16 # Uniform distribution
|
|
deviation = abs(count - expected_freq) / expected_freq
|
|
|
|
if deviation > 0.5: # 50% deviation from expected
|
|
anomalies.append(
|
|
f"Nibble 0x{ nibble_val:X}: {count} occurrences "
|
|
f"({count/total_nibbles*100:.1f}%), expected {expected_freq:.1f} "
|
|
f"(deviation: {deviation*100:.1f}%)"
|
|
)
|
|
|
|
return anomalies
|
|
|
|
def generate_improvement_suggestions(edges: List[ContainerEdge]) -> List[Dict[str, Any]]:
|
|
"""Generate improvement suggestions based on nibble-level analysis."""
|
|
print("\nGenerating improvement suggestions based on nibble-level analysis...")
|
|
|
|
suggestions = []
|
|
|
|
for edge in edges:
|
|
edge_suggestions = []
|
|
|
|
# Analyze nibble patterns
|
|
anomalies = detect_nibble_anomalies(edge.nibble_patterns)
|
|
|
|
if anomalies:
|
|
edge_suggestions.append({
|
|
"type": "nibble_distribution",
|
|
"severity": "medium",
|
|
"description": f"Unusual nibble distribution detected on {edge.edge_id}",
|
|
"anomalies": anomalies,
|
|
"suggestion": "Consider using compression or encryption to normalize data patterns"
|
|
})
|
|
|
|
# Analyze entropy
|
|
if edge.entropy < 2.0:
|
|
edge_suggestions.append({
|
|
"type": "low_entropy",
|
|
"severity": "low",
|
|
"description": f"Low entropy on {edge.edge_id}: {edge.entropy:.3f}",
|
|
"suggestion": "Data may be compressible or predictable, consider compression"
|
|
})
|
|
elif edge.entropy > 7.0:
|
|
edge_suggestions.append({
|
|
"type": "high_entropy",
|
|
"severity": "low",
|
|
"description": f"High entropy on {edge.edge_id}: {edge.entropy:.3f}",
|
|
"suggestion": "Data appears encrypted or highly random, verify integrity"
|
|
})
|
|
|
|
# Analyze data patterns
|
|
if len(edge.data) < 16:
|
|
edge_suggestions.append({
|
|
"type": "small_packet",
|
|
"severity": "low",
|
|
"description": f"Small packet on {edge.edge_id}: {len(edge.data)} bytes",
|
|
"suggestion": "Consider batching small packets to reduce overhead"
|
|
})
|
|
|
|
# Protocol-specific suggestions
|
|
if edge.protocol == "HTTP":
|
|
edge_suggestions.append({
|
|
"type": "protocol_upgrade",
|
|
"severity": "medium",
|
|
"description": f"HTTP protocol on {edge.edge_id}",
|
|
"suggestion": "Consider upgrading to HTTP/2 or HTTP/3 for better performance"
|
|
})
|
|
|
|
if edge_suggestions:
|
|
suggestions.append({
|
|
"edge_id": edge.edge_id,
|
|
"source": edge.source,
|
|
"destination": edge.destination,
|
|
"suggestions": edge_suggestions
|
|
})
|
|
|
|
print(f" Generated {len(suggestions)} improvement suggestion sets")
|
|
for suggestion_set in suggestions:
|
|
print(f" {suggestion_set['edge_id']}: {len(suggestion_set['suggestions'])} suggestions")
|
|
|
|
return suggestions
|
|
|
|
def execute_architect_deep_nibble_scan():
|
|
"""Execute deep nibble-level container scan on architect node."""
|
|
print("=" * 70)
|
|
print("Executing Deep Nibble-Level Container Scan on Architect Node")
|
|
print("=" * 70)
|
|
print("Configuration:")
|
|
print(" Target: architect node (Tailscale mesh)")
|
|
print(" Scan Level: Nibble (4-bit granularity)")
|
|
print(" Focus: Container edges and infrastructure")
|
|
print(" Goal: Generate improvement suggestions")
|
|
print("=" * 70)
|
|
|
|
# Connect to architect
|
|
architect_ip = connect_to_architect()
|
|
|
|
if not architect_ip:
|
|
print("Failed to connect to architect. Exiting.")
|
|
return None
|
|
|
|
# Scan container processes
|
|
processes = scan_container_processes(architect_ip)
|
|
|
|
if not processes:
|
|
print("No container processes found. Exiting.")
|
|
return None
|
|
|
|
# Scan container edges
|
|
edges = scan_container_edges(architect_ip)
|
|
|
|
if not edges:
|
|
print("No container edges found. Exiting.")
|
|
return None
|
|
|
|
# Analyze nibble patterns for each edge
|
|
print("\nAnalyzing nibble patterns for each edge...")
|
|
for edge in edges:
|
|
print(f"\n {edge.edge_id}:")
|
|
|
|
# Count nibble frequencies
|
|
nibble_counts = defaultdict(int)
|
|
for pattern in edge.nibble_patterns:
|
|
nibble_counts[pattern.value] += 1
|
|
|
|
print(f" Nibble distribution:")
|
|
for nibble_val in range(16):
|
|
count = nibble_counts.get(nibble_val, 0)
|
|
percentage = count / len(edge.nibble_patterns) * 100
|
|
bar = "█" * int(percentage / 2)
|
|
print(f" 0x{ nibble_val:X}: {count:3d} ({percentage:5.1f}%) {bar}")
|
|
|
|
# Detect anomalies
|
|
edge.anomalies = detect_nibble_anomalies(edge.nibble_patterns)
|
|
if edge.anomalies:
|
|
print(f" Anomalies detected: {len(edge.anomalies)}")
|
|
for anomaly in edge.anomalies:
|
|
print(f" ⚠️ {anomaly}")
|
|
else:
|
|
print(f" No anomalies detected")
|
|
|
|
# Generate improvement suggestions
|
|
suggestions = generate_improvement_suggestions(edges)
|
|
|
|
# Save results
|
|
results_path = f"shared-data/data/swarm_responses/architect_nibble_scan_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
|
Path(results_path).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
results = {
|
|
"scan_id": f"architect_nibble_scan_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
|
|
"timestamp": datetime.now().isoformat(),
|
|
"target_node": "architect",
|
|
"target_ip": architect_ip,
|
|
"scan_type": "deep_nibble_level",
|
|
"scan_granularity": "4_bits",
|
|
"processes": processes,
|
|
"edges": [
|
|
{
|
|
"edge_id": edge.edge_id,
|
|
"source": edge.source,
|
|
"destination": edge.destination,
|
|
"protocol": edge.protocol,
|
|
"data_length": len(edge.data),
|
|
"nibble_count": len(edge.nibble_patterns),
|
|
"entropy": edge.entropy,
|
|
"anomalies": edge.anomalies,
|
|
"nibble_distribution": {
|
|
f"0x{ i:X}": sum(1 for p in edge.nibble_patterns if p.value == i)
|
|
for i in range(16)
|
|
}
|
|
}
|
|
for edge in edges
|
|
],
|
|
"suggestions": suggestions,
|
|
"summary": {
|
|
"total_processes": len(processes),
|
|
"total_edges": len(edges),
|
|
"total_nibbles_analyzed": sum(len(edge.nibble_patterns) for edge in edges),
|
|
"total_anomalies": sum(len(edge.anomalies) for edge in edges),
|
|
"total_suggestions": sum(len(s['suggestions']) for s in suggestions)
|
|
}
|
|
}
|
|
|
|
with open(results_path, 'w') as f:
|
|
json.dump(results, f, indent=2)
|
|
|
|
print("\n" + "=" * 70)
|
|
print("Deep Nibble-Level Container Scan Complete")
|
|
print("=" * 70)
|
|
print(f"Total Processes: {results['summary']['total_processes']}")
|
|
print(f"Total Edges: {results['summary']['total_edges']}")
|
|
print(f"Total Nibbles Analyzed: {results['summary']['total_nibbles_analyzed']}")
|
|
print(f"Total Anomalies: {results['summary']['total_anomalies']}")
|
|
print(f"Total Suggestions: {results['summary']['total_suggestions']}")
|
|
print(f"\nResults saved to: {results_path}")
|
|
print("=" * 70)
|
|
|
|
return results
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
result = execute_architect_deep_nibble_scan()
|
|
if result:
|
|
print("\n✅ Deep nibble-level container scan completed")
|
|
print("\nArchitect node container edges analyzed at nibble level")
|
|
print("Improvement suggestions generated")
|
|
print("Results saved for review")
|
|
else:
|
|
print("\n❌ Failed to execute deep nibble-level container scan")
|
|
except Exception as e:
|
|
print(f"\n❌ Error: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|