Guides & Examples

# plugins/circuit_monitor.py
import time
from stem import Signal
from stem.control import Controller

class CircuitMonitor:
    def __init__(self, control_port=9051, password='your_password'):
        self.controller = Controller.from_port(port=control_port)
        self.controller.authenticate(password=password)
        self.circuit_times = {}
        
    def get_circuit_info(self):
        """Get current circuit information"""
        circuits = self.controller.get_circuits()
        circuit_info = []
        
        for circuit in circuits:
            if circuit.status == 'BUILT':
                path = [r.fingerprint for r in circuit.path]
                exit_fp = path[-1] if path else None
                exit_relay = self.controller.get_network_status(exit_fp)
                
                circuit_info.append({
                    'id': circuit.id,
                    'path_length': len(path),
                    'exit_country': exit_relay.address if exit_relay else 'Unknown',
                    'build_time': circuit.created,
                    'purpose': circuit.purpose
                })
                
        return circuit_info
        
    def rotate_circuit(self):
        """Force new Tor circuit"""
        self.controller.signal(Signal.NEWNYM)
        time.sleep(self.controller.get_newnym_wait())
        return self.get_circuit_info()
        
    def monitor_performance(self):
        """Track circuit performance metrics"""
        circuits = self.get_circuit_info()
        metrics = {
            'total_circuits': len(circuits),
            'countries': set(),
            'avg_path_length': 0
        }
        
        for circuit in circuits:
            metrics['countries'].add(circuit['exit_country'])
            metrics['avg_path_length'] += circuit['path_length']
            
        metrics['avg_path_length'] /= len(circuits) if circuits else 1
        metrics['countries'] = list(metrics['countries'])
        
        return metrics

# Advanced search with aggregations
POST /torscan_content/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "multi_match": {
            "query": "data breach credit card",
            "fields": ["content^2", "url.text", "title^3"],
            "type": "best_fields",
            "fuzziness": "AUTO"
          }
        }
      ],
      "filter": [
        {
          "range": {
            "found_at": {
              "gte": "now-7d/d"
            }
          }
        },
        {
          "terms": {
            "matched_patterns": ["credit_card", "personal_data"]
          }
        }
      ]
    }
  },
  "aggs": {
    "patterns_over_time": {
      "date_histogram": {
        "field": "found_at",
        "calendar_interval": "day"
      },
      "aggs": {
        "top_patterns": {
          "terms": {
            "field": "matched_patterns",
            "size": 5
          }
        }
      }
    },
    "ioc_types": {
      "nested": {
        "path": "iocs"
      },
      "aggs": {
        "by_type": {
          "terms": {
            "field": "iocs.type"
          }
        }
      }
    }
  },
  "highlight": {
    "fields": {
      "content": {
        "fragment_size": 150,
        "number_of_fragments": 3
      }
    }
  }
}

# config/opencti_connector.py
from pycti import OpenCTIApiClient
from datetime import datetime
import stix2

class OpenCTIConnector:
    def __init__(self, url, token):
        self.client = OpenCTIApiClient(url, token)
        
    def create_incident_from_scan(self, scan_data):
        """Create OpenCTI incident from scan results"""
        
        # Create incident
        incident = stix2.Incident(
            name=f"Dark Web Activity - {scan_data['name']}",
            description=self._build_description(scan_data),
            first_seen=scan_data['started_at'],
            last_seen=scan_data['completed_at'],
            confidence=85,
            labels=['dark-web', 'torscan', 'automated-detection']
        )
        
        # Create indicators
        indicators = []
        for result in scan_data['results']:
            for ioc in result['iocs']:
                indicator = stix2.Indicator(
                    pattern=self._create_pattern(ioc),
                    pattern_type='stix',
                    name=f"{ioc['type']}: {ioc['value']}",
                    description=f"Found on {result['url']}",
                    confidence=result['confidence'] * 100,
                    valid_from=result['found_at']
                )
                indicators.append(indicator)
                
        # Create relationships
        relationships = []
        for indicator in indicators:
            rel = stix2.Relationship(
                source_ref=incident.id,
                target_ref=indicator.id,
                relationship_type='indicates'
            )
            relationships.append(rel)
            
        # Bundle and submit
        bundle = stix2.Bundle(
            incident,
            *indicators,
            *relationships
        )
        
        response = self.client.stix2.import_bundle(bundle)
        return response