11.2.1 API Automation

import requests
import json

def get_threat_intel(api_key, indicator):
    """Ambil threat intelligence dari API eksternal."""
    url = f"https://otx.alienvault.com/api/v1/indicators/IPv4/{indicator}/general"

    headers = {
        "X-OTX-API-Key": api_key,
        "User-Agent": "SecurityBot/1.0"
    }

    try:
        response = requests.get(url, headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()

        print(f"[+] Threat Intel untuk {indicator}")
        print(f"    Pulse count: {data.get('pulse_info', {}).get('count', 0)}")
        print(f"    Reputation: {data.get('reputation', 'unknown')}")
        return data

    except requests.exceptions.Timeout:
        print(f"[!] Timeout saat request {indicator}")
    except requests.exceptions.HTTPError as e:
        print(f"[!] HTTP Error: {e.response.status_code} - {e.response.text}")
    except requests.exceptions.ConnectionError:
        print(f"[!] Gagal koneksi ke API")
    except json.JSONDecodeError:
        print(f"[!] Response bukan JSON valid")

    return None

# Demo
get_threat_intel("your_api_key", "8.8.8.8")

import requests
from datetime import datetime

class SIEMClient:
    """Client Python untuk mengirim alert ke SIEM via API."""

    def __init__(self, base_url, api_key):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "User-Agent": "SecurityBot/1.0"
        })

    def send_alert(self, title, severity, source_ip, description):
        """Kirim security alert ke SIEM."""
        payload = {
            "title": title,
            "severity": severity,        # critical, high, medium, low
            "source_ip": source_ip,
            "description": description,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "source": "automation-script"
        }

        try:
            response = self.session.post(
                f"{self.base_url}/api/alerts",
                json=payload,
                timeout=10
            )
            response.raise_for_status()
            print(f"[OK] Alert terkirim: {title}")
            return response.json()

        except requests.exceptions.RequestException as e:
            print(f"[FAIL] Gagal kirim alert: {e}")
            return None

# Demo
siem = SIEMClient(
    base_url="https://siem.company.com",
    api_key="sk-siem-abc123"
)

siem.send_alert(
    title="Brute Force Detected",
    severity="high",
    source_ip="10.0.0.5",
    description="100+ failed SSH attempts dalam 5 menit"
)

import requests

class FirewallAPI:
    """Automate firewall rule management via REST API."""

    def __init__(self, base_url, api_key):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "X-API-Key": api_key,
            "Content-Type": "application/json"
        })

    def block_ip(self, ip_address, reason="No reason"):
        """Blok IP address via firewall API."""
        payload = {
            "ip": ip_address,
            "action": "block",
            "protocol": "any",
            "reason": reason,
            "ttl": 86400  # 24 jam
        }

        response = self.session.post(
            f"{self.base_url}/api/rules",
            json=payload
        )

        if response.status_code == 201:
            print(f"[BLOCKED] {ip_address} - {reason}")
            return response.json()["id"]
        else:
            print(f"[FAIL] Gagal block {ip_address}: {response.text}")
            return None

    def unblock_ip(self, rule_id):
        """Hapus rule block."""
        response = self.session.delete(
            f"{self.base_url}/api/rules/{rule_id}"
        )

        if response.status_code == 200:
            print(f"[UNBLOCKED] Rule {rule_id} dihapus")
            return True
        else:
            print(f"[FAIL] Gagal unblock: {response.text}")
            return False

    def list_active_blocks(self):
        """List semua aturan block yang aktif."""
        response = self.session.get(
            f"{self.base_url}/api/rules?action=block&status=active"
        )

        if response.status_code == 200:
            rules = response.json()
            print(f"[INFO] {len(rules)} active block rules:")
            for rule in rules:
                print(f"  - {rule['ip']} (since {rule['created_at']})")
            return rules
        return []

# Demo
fw = FirewallAPI(
    base_url="https://firewall-api.company.com",
    api_key="fw-api-secret-456"
)

# Block attacker IP
rule_id = fw.block_ip("203.0.113.5", "SSH brute force detected")

# List active blocks
fw.list_active_blocks()

# Unblock jika sudah selesai insiden
# fw.unblock_ip(rule_id)

headers = {"X-API-Key": "abc123def456"}
response = requests.get("https://api.target.com/v1/data", headers=headers)

# Login dulu dapatkan token
login_resp = requests.post("https://api.target.com/auth/login",
    json={"username": "admin", "password": "secret"})

token = login_resp.json()["access_token"]

# Gunakan token untuk request selanjutnya
headers = {"Authorization": f"Bearer {token}"}
response = requests.get("https://api.target.com/api/reports", headers=headers)

import requests

class OAuth2Client:
    """Client dengan OAuth2 Client Credentials flow."""

    def __init__(self, token_url, client_id, client_secret):
        self.token_url = token_url
        self.client_id = client_id
        self.client_secret = client_secret
        self.access_token = None
        self.session = requests.Session()

    def authenticate(self):
        """Dapatkan access token."""
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        response = requests.post(self.token_url, data=payload)
        response.raise_for_status()

        data = response.json()
        self.access_token = data["access_token"]
        self.session.headers.update({
            "Authorization": f"Bearer {self.access_token}"
        })

        print(f"[AUTH] Token didapatkan, expires in {data.get('expires_in', 'N/A')}s")

    def get(self, endpoint):
        """GET request dengan auto-refresh."""
        if not self.access_token:
            self.authenticate()

        response = self.session.get(endpoint)
        if response.status_code == 401:
            print("[AUTH] Token expired, refresh...")
            self.authenticate()
            response = self.session.get(endpoint)

        return response

# Demo
client = OAuth2Client(
    token_url="https://auth.company.com/oauth/token",
    client_id="security-bot",
    client_secret="client-secret-789"
)

client.authenticate()
response = client.get("https://api.company.com/v2/threats")
print(response.json()[:2])  # Tampilkan 2 data pertama

import requests
import time

class PaginatedAPIClient:
    """Handle pagination untuk API dengan data besar."""

    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def fetch_all_logs(self, endpoint, max_pages=100):
        """Ambil semua data log dengan pagination."""
        all_data = []
        page = 1

        while page <= max_pages:
            params = {
                "page": page,
                "per_page": 100
            }

            try:
                response = requests.get(
                    f"{self.base_url}{endpoint}",
                    headers=self.headers,
                    params=params,
                    timeout=30
                )
                response.raise_for_status()

                data = response.json()
                if not data:
                    break  # Tidak ada data lagi

                all_data.extend(data)
                print(f"[PAGE {page}] Mendapatkan {len(data)} records")

                # Cek Link header untuk next page
                link_header = response.headers.get("Link", "")
                if 'rel="next"' not in link_header:
                    break  # Ini halaman terakhir

                page += 1

                # Rate limiting delay
                time.sleep(0.5)

            except requests.exceptions.RequestException as e:
                print(f"[FAIL] Error di page {page}: {e}")
                break

        print(f"[DONE] Total {len(all_data)} records dari {page} halaman")
        return all_data

# Demo
client = PaginatedAPIClient(
    base_url="https://siem.company.com",
    api_key="siem-token-xyz"
)

logs = client.fetch_all_logs("/api/logs", max_pages=5)

import time
import requests
from functools import wraps

def rate_limiter(max_calls=10, period=60):
    """Decorator: batasi jumlah request per periode waktu."""
    calls = []

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            # Hapus panggilan yang sudah kadaluarsa
            calls[:] = [c for c in calls if c > now - period]

            if len(calls) >= max_calls:
                wait_time = calls[0] + period - now
                print(f"[RATE LIMIT] Tunggu {wait_time:.1f}s...")
                time.sleep(wait_time)

            calls.append(time.time())
            return func(*args, **kwargs)
        return wrapper
    return decorator

class RateLimitedAPIClient:
    """API client dengan built-in rate limiting."""

    def __init__(self, base_url, api_key, max_rpm=60):
        self.base_url = base_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.max_rpm = max_rpm
        self.request_times = []

    def _wait_if_needed(self):
        """Tunggu jika接近 rate limit."""
        now = time.time()
        # Hanya hitung request dalam 1 menit terakhir
        self.request_times = [t for t in self.request_times if t > now - 60]

        if len(self.request_times) >= self.max_rpm:
            wait = self.request_times[0] + 60 - now
            print(f"[WAIT] Rate limit: tunggu {wait:.1f}s")
            time.sleep(wait + 0.5)

        self.request_times.append(time.time())

    def get(self, endpoint, params=None):
        """GET request dengan rate limiting otomatis."""
        self._wait_if_needed()

        response = requests.get(
            f"{self.base_url}{endpoint}",
            headers=self.headers,
            params=params
        )

        # Handle 429 Too Many Requests
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 30))
            print(f"[429] Rate limited! Tunggu {retry_after}s...")
            time.sleep(retry_after)
            return self.get(endpoint, params)  # Retry

        response.raise_for_status()
        return response.json()

# Demo
client = RateLimitedAPIClient(
    base_url="https://api.github.com",
    api_key="ghp_token123",
    max_rpm=30  # GitHub allows 60 unauthenticated, 5000 authenticated
)

# Simulasi batch request
repos = ["nousresearch/hermes-agent", "torvalds/linux", "curl/curl"]
for repo in repos:
    data = client.get(f"/repos/{repo}")
    print(f"[OK] {repo}: {data.get('stargazers_count')} stars")

import time
import random

def api_call_with_retry(url, headers, max_retries=5):
    """API call dengan exponential backoff."""
    for attempt in range(1, max_retries + 1):
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()
            return response.json()

        except requests.exceptions.RequestException as e:
            if attempt == max_retries:
                print(f"[FAIL] Gagal setelah {max_retries} percobaan")
                raise

            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"[RETRY] Attempt {attempt}/{max_retries} - tunggu {wait:.1f}s")
            print(f"        Error: {e}")
            time.sleep(wait)

#!/usr/bin/env python3
# auto-reporting.py
# Kirim laporan security harian ke SIEM dan notifikasi

import requests
import json
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText

class DailyReporter:
    """Generate dan kirim laporan security harian otomatis."""

    def __init__(self, siem_url, siem_key, webhook_url):
        self.siem_url = siem_url
        self.siem_key = siem_key
        self.webhook_url = webhook_url  # Slack/Telegram webhook

    def collect_data(self):
        """Kumpulkan data dari SIEM untuk laporan."""
        yesterday = datetime.utcnow() - timedelta(days=1)
        params = {
            "from": yesterday.isoformat() + "Z",
            "to": datetime.utcnow().isoformat() + "Z"
        }

        headers = {"Authorization": f"Bearer {self.siem_key}"}

        # Ambil jumlah alerts per severity
        stats = {}
        for severity in ["critical", "high", "medium", "low"]:
            params["severity"] = severity
            response = requests.get(
                f"{self.siem_url}/api/alerts/count",
                headers=headers, params=params
            )
            stats[severity] = response.json().get("count", 0)

        return stats

    def generate_report(self, stats):
        """Generate laporan dalam format JSON."""
        report = {
            "title": "Daily Security Report",
            "date": datetime.utcnow().strftime("%Y-%m-%d"),
            "summary": stats,
            "total_alerts": sum(stats.values()),
            "generated_at": datetime.utcnow().isoformat() + "Z"
        }
        report["status"] = "CRITICAL" if stats.get("critical", 0) > 0 else "NORMAL"

        return report

    def send_to_webhook(self, report):
        """Kirim laporan ke Slack/Telegram via webhook."""
        message = {
            "text": (
                f"*Daily Security Report*\n"
                f"Tanggal: {report['date']}\n"
                f"Status: {report['status']}\n"
                f"Total Alerts: {report['total_alerts']}\n"
                f"  - Critical: {report['summary'].get('critical', 0)}\n"
                f"  - High: {report['summary'].get('high', 0)}\n"
                f"  - Medium: {report['summary'].get('medium', 0)}\n"
                f"  - Low: {report['summary'].get('low', 0)}\n"
            )
        }

        response = requests.post(self.webhook_url, json=message)
        print(f"[WEBHOOK] Status: {response.status_code}")

    def run(self):
        """Jalankan pipeline reporting."""
        print("[REPORT] Mengumpulkan data...")
        stats = self.collect_data()

        print("[REPORT] Generate laporan...")
        report = self.generate_report(stats)

        print("[REPORT] Mengirim laporan...")
        self.send_to_webhook(report)

        print(f"[DONE] Laporan terkirim - {report['total_alerts']} total alerts")
        return report

# Run
if __name__ == "__main__":
    reporter = DailyReporter(
        siem_url="https://siem.company.com",
        siem_key="sk-siem-abc",
        webhook_url="https://hooks.slack.com/services/T00/B00/xxxx"
    )
    reporter.run()

#!/usr/bin/env python3
# siem-integration.py
# Ambil threat intelligence dari AlienVault OTX dan push ke SIEM

import requests
import time

class ThreatIntelPipeline:
    """Pipeline threat intelligence: OTX -> enrichment -> SIEM."""

    def __init__(self, otx_key, siem_url, siem_key):
        self.otx_key = otx_key
        self.siem_url = siem_url
        self.siem_key = siem_key

    def get_recent_pulses(self, limit=10):
        """Ambil pulse terbaru dari OTX."""
        url = "https://otx.alienvault.com/api/v1/pulses/subscribed"
        headers = {"X-OTX-API-Key": self.otx_key}

        response = requests.get(url, headers=headers,
                                params={"limit": limit, "page": 1})
        response.raise_for_status()
        return response.json().get("results", [])

    def extract_iocs(self, pulse):
        """Ekstrak Indicators of Compromise dari pulse."""
        iocs = []
        for indicator in pulse.get("indicators", []):
            iocs.append({
                "type": indicator.get("type"),
                "indicator": indicator.get("indicator"),
                "description": pulse.get("name", ""),
                "pulse_id": pulse.get("id"),
                "tlp": pulse.get("tlp", "amber"),
                "tags": [tag.get("name") for tag in pulse.get("tags", [])]
            })
        return iocs

    def push_to_siem(self, ioc):
        """Push indicator ke SIEM sebagai threat intel."""
        headers = {
            "Authorization": f"Bearer {self.siem_key}",
            "Content-Type": "application/json"
        }

        response = requests.post(
            f"{self.siem_url}/api/threat-intel",
            headers=headers, json=ioc, timeout=10
        )
        return response.status_code == 201

    def run(self):
        """Eksekusi pipeline end-to-end."""
        print("[THREAT INTEL] Mengambil pulses dari OTX...")
        pulses = self.get_recent_pulses(limit=5)

        total_iocs = 0
        for pulse in pulses:
            iocs = self.extract_iocs(pulse)
            for ioc in iocs:
                if self.push_to_siem(ioc):
                    total_iocs += 1
                time.sleep(0.2)  # Rate limiting

            print(f"  Pulse '{pulse.get('name', 'N/A')}': {len(iocs)} IOCs")

        print(f"[DONE] {total_iocs} indicators pushed ke SIEM")

# Run
if __name__ == "__main__":
    pipeline = ThreatIntelPipeline(
        otx_key="otx_key_here",
        siem_url="https://siem.company.com",
        siem_key="siem_key_here"
    )
    pipeline.run()

#!/usr/bin/env python3
# auto-block-firewall.py
# Deteksi brute force dari log dan auto-block via firewall API

import requests
import re
import time
from collections import Counter
from datetime import datetime, timedelta

class AutoBlockFirewall:
    """Baca log auth, deteksi brute force, auto-block via API firewall."""

    def __init__(self, log_file, firewall_api_url, firewall_api_key):
        self.log_file = log_file
        self.firewall = FirewallAPI(firewall_api_url, firewall_api_key)
        self.threshold = 10  # Jumlah failed attempts dalam window
        self.window_minutes = 15

    def parse_failed_attempts(self):
        """Parse failed SSH attempts dari auth.log."""
        pattern = r"Failed password for .* from (\d+\.\d+\.\d+\.\d+)"
        attempts = []

        try:
            with open(self.log_file, "r") as f:
                for line in f:
                    match = re.search(pattern, line)
                    if match:
                        attempts.append(match.group(1))
        except FileNotFoundError:
            print(f"[ERROR] Log file {self.log_file} tidak ditemukan")
            return []

        return attempts

    def detect_attackers(self):
        """Deteksi IP yang melebihi threshold brute force."""
        attempts = self.parse_failed_attempts()
        ip_counts = Counter(attempts)

        attackers = [
            {"ip": ip, "count": count}
            for ip, count in ip_counts.items()
            if count >= self.threshold
        ]

        attackers.sort(key=lambda x: x["count"], reverse=True)

        return attackers

    def auto_block(self, dry_run=True):
        """Auto-block IP attacker via firewall API."""
        print(f"[SCAN] Menganalisis {self.log_file}...")
        attackers = self.detect_attackers()

        if not attackers:
            print("[OK] Tidak ada attacker terdeteksi")
            return []

        print(f"[ALERT] Ditemukan {len(attackers)} IP mencurigakan:")
        blocked = []

        for attacker in attackers:
            print(f"  - {attacker['ip']}: {attacker['count']} attempts")

            if dry_run:
                print(f"    [DRY RUN] Akan di-block (skip)")
                continue

            rule_id = self.firewall.block_ip(
                ip_address=attacker["ip"],
                reason=f"Brute force: {attacker['count']} failed attempts"
            )

            if rule_id:
                blocked.append(attacker["ip"])
            time.sleep(0.5)  # Rate limit firewall API

        if dry_run:
            print(f"[INFO] Dry run mode - tidak ada yang benar-benar di-block")

        return blocked

# Demo - dry run dulu
blocker = AutoBlockFirewall(
    log_file="/var/log/auth.log",
    firewall_api_url="https://firewall-api.company.com",
    firewall_api_key="fw-secret-123"
)

# Test dengan dry_run=True dulu
blocker.auto_block(dry_run=True)

# Untuk production, set dry_run=False
# blocker.auto_block(dry_run=False)

import os
import requests
import logging
import time
from functools import wraps

# Konfigurasi dari environment variable
API_KEY = os.environ.get("API_KEY")
API_URL = os.environ.get("API_URL", "https://api.default.com")

# Logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

def safe_api_call(func):
    """Decorator: amankan semua API call dengan error handling."""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.Timeout:
            logger.error("Request timeout - API mungkin down")
        except requests.exceptions.HTTPError as e:
            logger.error(f"HTTP {e.response.status_code}: {e.response.text}")
        except requests.exceptions.ConnectionError:
            logger.error("Gagal koneksi - cek network/firewall")
        except Exception as e:
            logger.exception(f"Unexpected error: {e}")
        return None
    return wrapper