Scripting Essentials - Automating Your Security Workflow¶

#!/bin/bash
# This is a comment
echo "Hello, World!"

# Variables
NAME="Security Analyst"
echo "Welcome, $NAME"

# Command output
CURRENT_DATE=$(date)
echo "Today is: $CURRENT_DATE"

# Comment
Write-Host "Hello, World!"

# Variables
$Name = "Security Analyst"
Write-Host "Welcome, $Name"

# Command output
$CurrentDate = Get-Date
Write-Host "Today is: $CurrentDate"

#!/usr/bin/env python3
# This is a comment
print("Hello, World!")

# Variables
name = "Security Analyst"
print(f"Welcome, {name}")

# Import modules
import datetime
current_date = datetime.datetime.now()
print(f"Today is: {current_date}")

# Bash
TARGET="192.168.1.1"
PORT=80

# PowerShell
$Target = "192.168.1.1"
$Port = 80

# Python
target = "192.168.1.1"
port = 80

# Bash
if [ -f "/etc/passwd" ]; then
    echo "File exists"
else
    echo "File not found"
fi

# PowerShell
if (Test-Path "C:\Windows\System32") {
    Write-Host "Directory exists"
} else {
    Write-Host "Directory not found"
}

# Python
import os
if os.path.exists("/etc/passwd"):
    print("File exists")
else:
    print("File not found")

# Bash - count from 1 to 5
for i in {1..5}; do
    echo "Count: $i"
done

# PowerShell - count from 1 to 5
foreach ($i in 1..5) {
    Write-Host "Count: $i"
}

# Python - count from 1 to 5
for i in range(1, 6):
    print(f"Count: {i}")

# Bash function
greet_user() {
    echo "Hello, $1!"
}
greet_user "Alice"

# PowerShell function
function Greet-User {
    param($Name)
    Write-Host "Hello, $Name!"
}
Greet-User -Name "Alice"

# Python function
def greet_user(name):
    print(f"Hello, {name}!")
greet_user("Alice")

# Bash
content=$(cat file.txt)
echo "$content"

# PowerShell
$content = Get-Content file.txt
Write-Host $content

# Python
with open('file.txt', 'r') as f:
    content = f.read()
print(content)

# Bash
echo "Hello" > output.txt
echo "World" >> output.txt  # Append

# PowerShell
"Hello" | Set-Content output.txt
"World" | Add-Content output.txt  # Append

# Python
with open('output.txt', 'w') as f:
    f.write("Hello\n")
with open('output.txt', 'a') as f:
    f.write("World\n")

# List files
ls -la        # Bash
Get-ChildItem # PowerShell
import os; os.listdir()  # Python

# Copy files
cp file1.txt file2.txt        # Bash
Copy-Item file1.txt file2.txt # PowerShell
import shutil; shutil.copy('file1.txt', 'file2.txt')  # Python

# Ping test
ping google.com          # Bash/PowerShell
import os; os.system("ping google.com")  # Python

# Check ports
nc -zv google.com 80     # Bash (netcat)
Test-NetConnection google.com -Port 80  # PowerShell
import socket; socket.create_connection(('google.com', 80), timeout=1)  # Python

# Bash - check command success
if ping -c 1 google.com &> /dev/null; then
    echo "Network is up"
else
    echo "Network is down"
fi

# PowerShell - try/catch
try {
    Test-NetConnection google.com -Port 80 -ErrorAction Stop
    Write-Host "Port is open"
} catch {
    Write-Host "Port is closed"
}

# Python - try/except
try:
    import socket
    socket.create_connection(('google.com', 80), timeout=1)
    print("Port is open")
except:
    print("Port is closed")

# Bash - make executable and run
chmod +x script.sh
./script.sh

# PowerShell - run script
.\script.ps1

# Python - make executable and run
chmod +x script.py
./script.py

# Bash - access parameters
echo "First parameter: $1"
echo "Second parameter: $2"

# PowerShell - access parameters
param($Param1, $Param2)
Write-Host "First parameter: $Param1"
Write-Host "Second parameter: $Param2"

# Python - access parameters
import sys
print(f"First parameter: {sys.argv[1]}")
print(f"Second parameter: {sys.argv[2]}")

# Bash - using grep with regex
echo "192.168.1.1" | grep -E "^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$"

# Python - regex module
import re
ip_pattern = r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"
if re.match(ip_pattern, "192.168.1.1"):
    print("Valid IP")

# Email validation
email_pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"

# URL validation
url_pattern = r"^https?://[^\s/$.?#].[^\s]*$"

# Phone number
phone_pattern = r"^\+?1?[-.\s]?\(?([0-9]{3})\)?[-.\s]?([0-9]{3})[-.\s]?([0-9]{4})$"

# Bash - process file line by line
while IFS= read -r line; do
    echo "Processing: $line"
done < large_file.txt

# Python - efficient file reading
with open('large_file.txt', 'r') as f:
    for line in f:
        process_line(line.strip())

import csv

# Reading CSV
with open('data.csv', 'r') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row['name'], row['email'])

# Writing CSV
with open('output.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['Name', 'Email'])
    writer.writerow(['John Doe', 'john@example.com'])

import json

# Reading JSON
with open('data.json', 'r') as f:
    data = json.load(f)
    print(data['users'][0]['name'])

# Writing JSON
user_data = {'name': 'John', 'age': 30}
with open('user.json', 'w') as f:
    json.dump(user_data, f, indent=2)

import requests
from bs4 import BeautifulSoup

def scrape_website(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')

    # Find all links
    links = soup.find_all('a')
    for link in links:
        print(link.get('href'))

    # Find specific elements
    titles = soup.find_all('h2', class_='title')
    for title in titles:
        print(title.text.strip())

# Usage
scrape_website('https://example.com')

import requests

# GET request
response = requests.get('https://api.github.com/user', auth=('user', 'pass'))
print(response.json())

# POST request
data = {'key': 'value'}
response = requests.post('https://httpbin.org/post', json=data)
print(response.json())

# Handling errors
try:
    response = requests.get('https://api.example.com/data')
    response.raise_for_status()  # Raise exception for bad status codes
    data = response.json()
except requests.exceptions.RequestException as e:
    print(f"Error: {e}")

import sqlite3

# Connect to database
conn = sqlite3.connect('security.db')
cursor = conn.cursor()

# Create table
cursor.execute('''
    CREATE TABLE IF NOT EXISTS scans (
        id INTEGER PRIMARY KEY,
        target TEXT,
        ports TEXT,
        timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
    )
''')

# Insert data
cursor.execute("INSERT INTO scans (target, ports) VALUES (?, ?)", 
               ('192.168.1.1', '22,80,443'))

# Query data
cursor.execute("SELECT * FROM scans WHERE target = ?", ('192.168.1.1',))
results = cursor.fetchall()

conn.commit()
conn.close()

import mysql.connector

# Connect to MySQL
conn = mysql.connector.connect(
    host='localhost',
    user='root',
    password='password',
    database='security'
)

cursor = conn.cursor()

# Create table
cursor.execute('''
    CREATE TABLE IF NOT EXISTS vulnerabilities (
        id INT AUTO_INCREMENT PRIMARY KEY,
        cve_id VARCHAR(50),
        severity VARCHAR(20),
        description TEXT
    )
''')

conn.commit()
conn.close()

import threading
import time

def scan_port(host, port):
    # Port scanning function
    print(f"Scanning {host}:{port}")

# Create threads
threads = []
for port in range(1, 1025):
    thread = threading.Thread(target=scan_port, args=('192.168.1.1', port))
    threads.append(thread)
    thread.start()

# Wait for all threads to complete
for thread in threads:
    thread.join()

import concurrent.futures
import requests

def check_website(url):
    try:
        response = requests.get(url, timeout=5)
        return f"{url}: {response.status_code}"
    except:
        return f"{url}: Error"

# Using ThreadPoolExecutor
urls = ['https://google.com', 'https://github.com', 'https://stackoverflow.com']
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(check_website, urls)
    for result in results:
        print(result)

import logging

# Configure logging
logging.basicConfig(
    filename='security.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

# Log messages
logging.info("Scan started")
logging.warning("Port 22 open on target")
logging.error("Connection failed")

# Custom logger
logger = logging.getLogger('security_scanner')
logger.setLevel(logging.DEBUG)

# File handler
file_handler = logging.FileHandler('scanner.log')
file_handler.setLevel(logging.DEBUG)

# Formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)

logger.addHandler(file_handler)

#!/bin/bash
LOGFILE="script.log"

log() {
    echo "$(date '+%Y-%m-%d %H:%M:%S') - $1" >> "$LOGFILE"
}

# Usage
log "Script started"
log "Processing target: $TARGET"
log "Script completed"

import configparser

config = configparser.ConfigParser()
config.read('config.ini')

# Read values
api_key = config['API']['key']
timeout = config['SETTINGS'].getint('timeout')

# Write values
config['NEW_SECTION'] = {'setting': 'value'}
with open('config.ini', 'w') as f:
    config.write(f)

import yaml

# Read YAML
with open('config.yaml', 'r') as f:
    config = yaml.safe_load(f)

print(config['database']['host'])

# Write YAML
data = {
    'database': {
        'host': 'localhost',
        'port': 5432
    }
}

with open('config.yaml', 'w') as f:
    yaml.dump(data, f, default_flow_style=False)

try:
    # Risky operation
    result = 10 / 0
except ZeroDivisionError as e:
    print(f"Division by zero: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")
finally:
    print("Cleanup code always runs")

# Custom exceptions
class SecurityError(Exception):
    pass

def scan_target(target):
    if not target:
        raise SecurityError("Target cannot be empty")
    # Scanning logic

#!/bin/bash
set -e  # Exit on any error
set -u  # Exit on undefined variables

# Error handling function
error_exit() {
    echo "Error: $1" >&2
    exit 1
}

# Usage
command_that_might_fail || error_exit "Command failed"

# Trap signals
trap 'error_exit "Script interrupted"' INT TERM

import unittest

def add_numbers(a, b):
    return a + b

class TestMath(unittest.TestCase):
    def test_add_positive(self):
        self.assertEqual(add_numbers(2, 3), 5)

    def test_add_negative(self):
        self.assertEqual(add_numbers(-1, 1), 0)

    def test_add_zero(self):
        self.assertEqual(add_numbers(0, 0), 0)

if __name__ == '__main__':
    unittest.main()

#!/bin/bash

test_add() {
    result=$(add 2 3)
    if [ "$result" -eq 5 ]; then
        echo "Test passed"
    else
        echo "Test failed"
        exit 1
    fi
}

# Run tests
test_add
echo "All tests passed"

#!/bin/bash
# Script: port_scanner.sh
# Purpose: Scan target host for open ports and identify services
# Author: Your Name
# Last Modified: 2024-01-15
# Usage: ./port_scanner.sh <target_ip> [port_range]
# Example: ./port_scanner.sh 192.168.1.1 1-1000

# Check if target provided
if [ -z "$1" ]; then
    echo "Error: Target IP required"
    echo "Usage: $0 <target_ip> [port_range]"
    exit 1
fi

TARGET="$1"
PORT_RANGE="${2:-1-1024}"  # Default to common ports if not specified

# Perform scan
# Using nmap with service detection for accuracy
nmap -sS -sV -p "$PORT_RANGE" "$TARGET"

#!/bin/bash
# script
TARGET=$1
# scan
nmap $TARGET

#!/bin/bash
# validate_inputs.sh

# Function to validate IP address
validate_ip() {
    local ip="$1"
    if [[ $ip =~ ^[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}$ ]]; then
        IFS='.' read -ra ADDR <<< "$ip"
        for i in "${ADDR[@]}"; do
            if [[ $i -gt 255 ]]; then
                return 1
            fi
        done
        return 0
    fi
    return 1
}

# Function to validate port number
validate_port() {
    local port="$1"
    if [[ "$port" =~ ^[0-9]+$ ]] && [ "$port" -ge 1 ] && [ "$port" -le 65535 ]; then
        return 0
    fi
    return 1
}

# Main script
if [ $# -lt 2 ]; then
    echo "Error: Missing required parameters"
    echo "Usage: $0 <target_ip> <port>"
    exit 1
fi

TARGET_IP="$1"
PORT="$2"

# Validate inputs
if ! validate_ip "$TARGET_IP"; then
    echo "Error: Invalid IP address: $TARGET_IP"
    exit 1
fi

if ! validate_port "$PORT"; then
    echo "Error: Invalid port number: $PORT (must be 1-65535)"
    exit 1
fi

echo "Scanning $TARGET_IP:$PORT"
# Continue with actual work...

import re
import sys

def validate_ip(ip):
    """Validate IPv4 address format."""
    pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
    if not re.match(pattern, ip):
        return False
    parts = ip.split('.')
    return all(0 <= int(part) <= 255 for part in parts)

def validate_port(port):
    """Validate port number (1-65535)."""
    try:
        port_num = int(port)
        return 1 <= port_num <= 65535
    except ValueError:
        return False

# Usage
if len(sys.argv) < 3:
    print("Error: Missing parameters")
    print(f"Usage: {sys.argv[0]} <ip> <port>")
    sys.exit(1)

target_ip = sys.argv[1]
port = sys.argv[2]

if not validate_ip(target_ip):
    print(f"Error: Invalid IP address: {target_ip}")
    sys.exit(1)

if not validate_port(port):
    print(f"Error: Invalid port: {port}")
    sys.exit(1)

TARGET_IP="192.168.1.1"
SCAN_TIMEOUT=5
OUTPUT_FILE="scan_results.txt"
MAX_RETRIES=3
ENABLE_VERBOSE=true

ip="192.168.1.1"
t=5
f="scan_results.txt"
r=3
v=true

#!/bin/bash
set -euo pipefail  # Exit on error, undefined vars, pipe failures

# Error handler function
error_exit() {
    echo "ERROR: $1" >&2
    exit "${2:-1}"
}

# Trap for cleanup on exit
cleanup() {
    # Remove temporary files
    rm -f /tmp/temp_*
    echo "Cleanup completed"
}
trap cleanup EXIT

# Use error_exit instead of letting script crash
if ! command -v nmap &> /dev/null; then
    error_exit "nmap not found. Please install nmap."
fi

# Check file exists before reading
CONFIG_FILE="${CONFIG_FILE:-config.txt}"
if [ ! -f "$CONFIG_FILE" ]; then
    error_exit "Config file not found: $CONFIG_FILE"
fi

# Try to read config, handle errors
if ! source "$CONFIG_FILE"; then
    error_exit "Failed to load config file: $CONFIG_FILE"
fi

import sys
import logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def read_config(config_file):
    """Read configuration file with error handling."""
    try:
        with open(config_file, 'r') as f:
            config = {}
            for line in f:
                if '=' in line and not line.strip().startswith('#'):
                    key, value = line.strip().split('=', 1)
                    config[key] = value
            return config
    except FileNotFoundError:
        logger.error(f"Config file not found: {config_file}")
        sys.exit(1)
    except PermissionError:
        logger.error(f"Permission denied reading: {config_file}")
        sys.exit(1)
    except Exception as e:
        logger.error(f"Error reading config: {e}")
        sys.exit(1)

# Usage
try:
    config = read_config('config.txt')
    target = config.get('target')
    if not target:
        raise ValueError("Target not specified in config")
except Exception as e:
    logger.error(f"Configuration error: {e}")
    sys.exit(1)

#!/bin/bash
# test_scanner.sh - Test script for port scanner

# Test 1: Valid input
echo "Test 1: Valid IP and port"
./port_scanner.sh 192.168.1.1 80
if [ $? -eq 0 ]; then
    echo "✓ Pass"
else
    echo "✗ Fail"
fi

# Test 2: Missing IP
echo "Test 2: Missing IP (should fail)"
./port_scanner.sh
if [ $? -ne 0 ]; then
    echo "✓ Pass (correctly failed)"
else
    echo "✗ Fail (should have failed)"
fi

# Test 3: Invalid IP
echo "Test 3: Invalid IP format"
./port_scanner.sh 999.999.999.999 80
if [ $? -ne 0 ]; then
    echo "✓ Pass"
else
    echo "✗ Fail"
fi

# Test 4: Invalid port
echo "Test 4: Invalid port (too high)"
./port_scanner.sh 192.168.1.1 99999
if [ $? -ne 0 ]; then
    echo "✓ Pass"
else
    echo "✗ Fail"
fi

# Test 5: Empty input
echo "Test 5: Empty string"
./port_scanner.sh "" 80
if [ $? -ne 0 ]; then
    echo "✓ Pass"
else
    echo "✗ Fail"
fi

import unittest
from scanner import validate_ip, validate_port, scan_port

class TestScanner(unittest.TestCase):
    def test_valid_ip(self):
        self.assertTrue(validate_ip("192.168.1.1"))
        self.assertTrue(validate_ip("8.8.8.8"))

    def test_invalid_ip(self):
        self.assertFalse(validate_ip("999.999.999.999"))
        self.assertFalse(validate_ip("not.an.ip"))
        self.assertFalse(validate_ip(""))
        self.assertFalse(validate_ip("192.168.1"))

    def test_valid_port(self):
        self.assertTrue(validate_port("80"))
        self.assertTrue(validate_port("65535"))
        self.assertTrue(validate_port("1"))

    def test_invalid_port(self):
        self.assertFalse(validate_port("0"))
        self.assertFalse(validate_port("65536"))
        self.assertFalse(validate_port("-1"))
        self.assertFalse(validate_port("abc"))

    def test_scan_port(self):
        # Test scanning localhost
        result = scan_port("127.0.0.1", 22)
        # Should return boolean or port status
        self.assertIsInstance(result, bool)

if __name__ == '__main__':
    unittest.main()

# config.ini or config.conf
TARGET_IP=192.168.1.1
SCAN_TIMEOUT=5
OUTPUT_DIR=/tmp/scans
WORDLIST=/usr/share/wordlists/rockyou.txt
THREADS=10
VERBOSE=true

# Load config
if [ -f "config.ini" ]; then
    source config.ini
else
    echo "Warning: config.ini not found, using defaults"
    TARGET_IP=""
    SCAN_TIMEOUT=5
fi

#!/bin/bash
# Use getopts for proper argument parsing

usage() {
    echo "Usage: $0 -t <target> -p <port> [-v] [-o <output>]"
    exit 1
}

TARGET=""
PORT=""
VERBOSE=false
OUTPUT=""

while getopts "t:p:vo:" opt; do
    case $opt in
        t) TARGET="$OPTARG" ;;
        p) PORT="$OPTARG" ;;
        v) VERBOSE=true ;;
        o) OUTPUT="$OPTARG" ;;
        *) usage ;;
    esac
done

if [ -z "$TARGET" ] || [ -z "$PORT" ]; then
    usage
fi

# Use variables instead of hardcoded values
if [ "$VERBOSE" = true ]; then
    echo "Scanning $TARGET:$PORT"
fi

import argparse

def main():
    parser = argparse.ArgumentParser(
        description='Port scanner for security testing',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog='''
Examples:
  %(prog)s -t 192.168.1.1 -p 80
  %(prog)s -t 192.168.1.1 -p 1-1000 -v
  %(prog)s -t 192.168.1.1 -p 80 -o results.txt
        '''
    )

    parser.add_argument('-t', '--target', required=True,
                       help='Target IP address or hostname')
    parser.add_argument('-p', '--port', required=True,
                       help='Port or port range (e.g., 80 or 1-1000)')
    parser.add_argument('-v', '--verbose', action='store_true',
                       help='Verbose output')
    parser.add_argument('-o', '--output',
                       help='Output file (default: stdout)')
    parser.add_argument('-T', '--timeout', type=int, default=5,
                       help='Timeout in seconds (default: 5)')
    parser.add_argument('-t', '--threads', type=int, default=10,
                       help='Number of threads (default: 10)')

    args = parser.parse_args()

    # Use args.target, args.port, etc.
    print(f"Scanning {args.target}:{args.port}")

if __name__ == '__main__':
    main()

#!/bin/bash

# Logging function
log() {
    local level="$1"
    shift
    local message="$*"
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')

    case "$level" in
        INFO)
            echo "[$timestamp] [INFO] $message" | tee -a "$LOG_FILE"
            ;;
        ERROR)
            echo "[$timestamp] [ERROR] $message" >&2 | tee -a "$LOG_FILE"
            ;;
        WARNING)
            echo "[$timestamp] [WARNING] $message" | tee -a "$LOG_FILE"
            ;;
        DEBUG)
            if [ "${DEBUG:-false}" = "true" ]; then
                echo "[$timestamp] [DEBUG] $message" | tee -a "$LOG_FILE"
            fi
            ;;
    esac
}

# Usage
LOG_FILE="script.log"
log INFO "Starting scan"
log DEBUG "Target: $TARGET"
log ERROR "Connection failed"

import logging
import sys

def setup_logging(verbose=False, log_file=None):
    """Configure logging for script."""
    level = logging.DEBUG if verbose else logging.INFO

    handlers = [logging.StreamHandler(sys.stdout)]
    if log_file:
        handlers.append(logging.FileHandler(log_file))

    logging.basicConfig(
        level=level,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=handlers
    )

    return logging.getLogger(__name__)

# Usage
logger = setup_logging(verbose=True, log_file='scan.log')
logger.info("Starting port scan")
logger.debug(f"Target: {target}, Ports: {ports}")
logger.error("Scan failed")

# BAD - Command injection vulnerability
TARGET=$1
nmap $TARGET  # If TARGET contains "; rm -rf /", you're screwed

# GOOD - Safe
TARGET="$1"
# Validate input first
if ! validate_ip "$TARGET"; then
    exit 1
fi
nmap "$TARGET"  # Quotes prevent injection

# BETTER - Use arrays
ARGS=("$TARGET")
nmap "${ARGS[@]}"

import subprocess

# BAD - Shell injection
target = input("Enter target: ")
subprocess.call(f"nmap {target}", shell=True)  # Dangerous!

# GOOD - No shell, list arguments
target = input("Enter target: ")
# Validate first
if not validate_ip(target):
    print("Invalid IP")
    sys.exit(1)

subprocess.call(["nmap", target])  # Safe

# BETTER - With error handling
try:
    result = subprocess.run(
        ["nmap", "-sS", target],
        capture_output=True,
        text=True,
        timeout=60,
        check=True
    )
    print(result.stdout)
except subprocess.TimeoutExpired:
    print("Scan timed out")
except subprocess.CalledProcessError as e:
    print(f"Scan failed: {e}")

# Create temp files securely
TEMP_FILE=$(mktemp)
chmod 600 "$TEMP_FILE"  # Read/write for owner only
# Use file
rm "$TEMP_FILE"  # Clean up

# DON'T put passwords in scripts
# BAD:
PASSWORD="mysecretpassword"
mysql -u root -p"$PASSWORD"

# GOOD: Use environment variables or config files with proper permissions
# .env file (chmod 600)
source .env
mysql -u root -p"$DB_PASSWORD"

# BETTER: Use secret management
# AWS Secrets Manager, HashiCorp Vault, etc.

#!/bin/bash

# Sequential (slow)
for port in {1..1024}; do
    nc -zv target.com $port
done

# Parallel (fast)
for port in {1..1024}; do
    (nc -zv target.com $port 2>&1 | grep open && echo "Port $port is open") &
done
wait  # Wait for all background jobs

# Limit parallelism
NUM_JOBS=0
MAX_JOBS=10

for port in {1..1024}; do
    # Wait if too many jobs running
    while [ $NUM_JOBS -ge $MAX_JOBS ]; do
        sleep 0.1
        NUM_JOBS=$(jobs -r | wc -l)
    done

    (nc -zv target.com $port 2>&1 | grep open && echo "Port $port is open") &
    ((NUM_JOBS++))
done
wait

import concurrent.futures
import socket

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

# Sequential (slow)
open_ports = []
for port in range(1, 1025):
    if scan_port("target.com", port):
        open_ports.append(port)

# Threading (faster)
open_ports = []
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
    futures = {executor.submit(scan_port, "target.com", port): port 
               for port in range(1, 1025)}
    for future in concurrent.futures.as_completed(futures):
        result = future.result()
        if result:
            open_ports.append(result)

# Port Scanner Script

## Description
Scans target host for open ports and identifies running services.

## Requirements
- nmap
- Bash 4.0+
- Root privileges (for some scan types)

## Installation
```bash
chmod +x port_scanner.sh

./port_scanner.sh <target_ip> [port_range]

# Examples
./port_scanner.sh 192.168.1.1
./port_scanner.sh 192.168.1.1 1-1000
./port_scanner.sh 192.168.1.1 80,443,8080

# Basic scan
./port_scanner.sh 192.168.1.1

# Scan specific range
./port_scanner.sh 192.168.1.1 1-1000

# Verbose output to file
./port_scanner.sh 192.168.1.1 -v -o results.txt

## Real-World Scripting Examples

Let's look at complete, production-ready scripts that actually solve real problems.

### Example 1: Automated Reconnaissance Script

This is a script I actually use. It's not perfect, but it works reliably.

```bash
#!/bin/bash
# recon.sh - Automated reconnaissance script
# Usage: ./recon.sh target.com

set -euo pipefail

# Colors for output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color

# Logging
log() {
    echo -e "${GREEN}[+]${NC} $1"
}

error() {
    echo -e "${RED}[-]${NC} $1" >&2
}

warning() {
    echo -e "${YELLOW}[!]${NC} $1"
}

# Validate input
if [ $# -lt 1 ]; then
    error "Usage: $0 <target>"
    exit 1
fi

TARGET="$1"
OUTPUT_DIR="recon_$(date +%Y%m%d_%H%M%S)_${TARGET}"
mkdir -p "$OUTPUT_DIR"

log "Starting reconnaissance for $TARGET"
log "Output directory: $OUTPUT_DIR"

# DNS enumeration
log "DNS enumeration..."
dig "$TARGET" any > "$OUTPUT_DIR/dns_any.txt" 2>&1
dig "$TARGET" MX >> "$OUTPUT_DIR/dns_mx.txt" 2>&1
dig "$TARGET" NS >> "$OUTPUT_DIR/dns_ns.txt" 2>&1

# Subdomain enumeration (if subfinder installed)
if command -v subfinder &> /dev/null; then
    log "Subdomain enumeration..."
    subfinder -d "$TARGET" > "$OUTPUT_DIR/subdomains.txt" 2>&1
else
    warning "subfinder not found, skipping subdomain enumeration"
fi

# Port scanning
log "Port scanning..."
if command -v nmap &> /dev/null; then
    nmap -sS -sV -O -oN "$OUTPUT_DIR/nmap_scan.txt" "$TARGET"
else
    error "nmap not found"
fi

# Web technologies
log "Web technology detection..."
if command -v whatweb &> /dev/null; then
    whatweb "$TARGET" > "$OUTPUT_DIR/whatweb.txt" 2>&1
fi

# Directory brute force
log "Directory enumeration..."
if command -v gobuster &> /dev/null; then
    gobuster dir -u "http://$TARGET" \
        -w /usr/share/wordlists/dirbuster/directory-list-2.3-medium.txt \
        -o "$OUTPUT_DIR/gobuster.txt" 2>&1
fi

log "Reconnaissance complete!"
log "Results in: $OUTPUT_DIR"

#!/usr/bin/env python3
"""
log_analyzer.py - Analyze security logs for suspicious activity

Usage:
    python log_analyzer.py <log_file> [options]

Options:
    --failed-logins    Analyze failed login attempts
    --suspicious-ips   Find suspicious IP addresses
    --time-window HOURS   Analyze last N hours
    --output FILE     Save results to file
"""

import re
import sys
import argparse
from collections import Counter
from datetime import datetime, timedelta

def analyze_failed_logins(log_file, hours=24):
    """Analyze failed login attempts."""
    cutoff = datetime.now() - timedelta(hours=hours)
    failed_logins = []

    with open(log_file, 'r') as f:
        for line in f:
            # Match failed login patterns
            if re.search(r'Failed|Invalid|Authentication failure', line, re.I):
                # Extract IP address
                ip_match = re.search(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', line)
                if ip_match:
                    failed_logins.append({
                        'ip': ip_match.group(),
                        'line': line.strip(),
                        'time': extract_timestamp(line)
                    })

    # Count by IP
    ip_counts = Counter(login['ip'] for login in failed_logins)

    print("\n=== Failed Login Analysis ===")
    print(f"Total failed attempts: {len(failed_logins)}")
    print(f"\nTop 10 attacking IPs:")
    for ip, count in ip_counts.most_common(10):
        print(f"  {ip}: {count} attempts")

    return failed_logins

def extract_timestamp(line):
    """Extract timestamp from log line."""
    # Common log formats
    patterns = [
        r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})',
        r'(\w{3} \d{1,2} \d{2}:\d{2}:\d{2})',
    ]

    for pattern in patterns:
        match = re.search(pattern, line)
        if match:
            return match.group(1)
    return None

def find_suspicious_ips(log_file):
    """Find IPs with suspicious activity patterns."""
    ip_activity = Counter()

    with open(log_file, 'r') as f:
        for line in f:
            # Multiple failed attempts from same IP
            if 'Failed' in line or 'Invalid' in line:
                ip_match = re.search(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', line)
                if ip_match:
                    ip_activity[ip_match.group()] += 1

    # IPs with > 10 failed attempts
    suspicious = {ip: count for ip, count in ip_activity.items() if count > 10}

    print("\n=== Suspicious IP Addresses ===")
    for ip, count in sorted(suspicious.items(), key=lambda x: x[1], reverse=True):
        print(f"  {ip}: {count} failed attempts")

    return suspicious

def main():
    parser = argparse.ArgumentParser(description='Analyze security logs')
    parser.add_argument('log_file', help='Path to log file')
    parser.add_argument('--failed-logins', action='store_true',
                       help='Analyze failed login attempts')
    parser.add_argument('--suspicious-ips', action='store_true',
                       help='Find suspicious IP addresses')
    parser.add_argument('--time-window', type=int, default=24,
                       help='Time window in hours (default: 24)')
    parser.add_argument('--output', help='Save results to file')

    args = parser.parse_args()

    try:
        if args.failed_logins:
            results = analyze_failed_logins(args.log_file, args.time_window)
        if args.suspicious_ips:
            results = find_suspicious_ips(args.log_file)

        if args.output:
            with open(args.output, 'w') as f:
                # Write results
                pass
            print(f"\nResults saved to: {args.output}")
    except FileNotFoundError:
        print(f"Error: Log file not found: {args.log_file}")
        sys.exit(1)
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

if __name__ == '__main__':
    main()

#!/bin/bash
# crack_hashes.sh - Automated hash cracking workflow

set -euo pipefail

HASH_FILE="$1"
HASH_TYPE="${2:-auto}"

if [ ! -f "$HASH_FILE" ]; then
    echo "Error: Hash file not found: $HASH_FILE"
    exit 1
fi

# Identify hash type if not specified
if [ "$HASH_TYPE" = "auto" ]; then
    echo "Identifying hash type..."
    HASH_TYPE=$(hash-identifier < "$HASH_FILE" | grep -i "possible hash" | head -1 | awk '{print $NF}')
    echo "Detected hash type: $HASH_TYPE"
fi

# Map hash types to hashcat mode
case "$HASH_TYPE" in
    MD5) MODE=0 ;;
    SHA1) MODE=100 ;;
    NTLM) MODE=1000 ;;
    SHA256) MODE=1400 ;;
    *) echo "Unknown hash type: $HASH_TYPE"; exit 1 ;;
esac

# Wordlists to try (in order)
WORDLISTS=(
    "/usr/share/wordlists/rockyou.txt"
    "/usr/share/wordlists/passwords/500-worst-passwords.txt"
    "/usr/share/wordlists/fasttrack.txt"
)

# Try each wordlist
for wordlist in "${WORDLISTS[@]}"; do
    if [ ! -f "$wordlist" ]; then
        continue
    fi

    echo "Trying wordlist: $wordlist"
    hashcat -m "$MODE" -a 0 "$HASH_FILE" "$wordlist" || true

    # Check if we cracked anything
    if hashcat -m "$MODE" --show "$HASH_FILE" | grep -q .; then
        echo "Success! Cracked hashes:"
        hashcat -m "$MODE" --show "$HASH_FILE"
        break
    fi
done

# Show results
hashcat -m "$MODE" --show "$HASH_FILE"

import queue
import threading
import time

def producer(q, data_source):
    """Generate data and put it in queue."""
    for item in data_source:
        q.put(item)
        time.sleep(0.1)  # Simulate processing delay
    q.put(None)  # Signal completion

def consumer(q, processor_func):
    """Consume data from queue and process."""
    while True:
        item = q.get()
        if item is None:
            break
        processor_func(item)
        q.task_done()

# Usage
q = queue.Queue(maxsize=100)  # Limit queue size
data = range(1000)

producer_thread = threading.Thread(target=producer, args=(q, data))
consumer_thread = threading.Thread(target=consumer, args=(q, process_item))

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()

import time
import random

def retry(func, max_attempts=3, delay=1, backoff=2):
    """Retry function with exponential backoff."""
    for attempt in range(max_attempts):
        try:
            return func()
        except Exception as e:
            if attempt == max_attempts - 1:
                raise
            wait_time = delay * (backoff ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time:.2f}s...")
            time.sleep(wait_time)

# Usage
def unreliable_network_request():
    if random.random() < 0.7:
        raise ConnectionError("Network error")
    return "Success"

result = retry(unreliable_network_request, max_attempts=5)

class Pipeline:
    def __init__(self):
        self.steps = []

    def add_step(self, func):
        self.steps.append(func)
        return self

    def execute(self, data):
        result = data
        for step in self.steps:
            result = step(result)
        return result

# Usage
pipeline = (Pipeline()
    .add_step(lambda x: x.strip())
    .add_step(lambda x: x.lower())
    .add_step(lambda x: x.replace(' ', '-'))
)

result = pipeline.execute("  HELLO WORLD  ")
# Result: "hello-world"

import requests
import time
from functools import wraps

class APIClient:
    def __init__(self, base_url, api_key=None, rate_limit=60):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        if api_key:
            self.session.headers.update({'Authorization': f'Bearer {api_key}'})
        self.rate_limit = rate_limit
        self.last_request = 0

    def rate_limit_decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            elapsed = time.time() - self.last_request
            if elapsed < (60 / self.rate_limit):
                time.sleep((60 / self.rate_limit) - elapsed)
            result = func(self, *args, **kwargs)
            self.last_request = time.time()
            return result
        return wrapper

    def get(self, endpoint, params=None):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()

# Usage
client = APIClient("https://api.virustotal.com/v2", api_key="YOUR_KEY")
result = client.get("/ip_address/report", params={"ip": "8.8.8.8"})

def process_large_file(filename, processor_func, chunk_size=1000):
    """Process large file in chunks."""
    buffer = []

    with open(filename, 'r') as f:
        for line in f:
            buffer.append(line.strip())

            if len(buffer) >= chunk_size:
                # Process chunk
                for item in buffer:
                    processor_func(item)
                buffer = []

        # Process remaining items
        for item in buffer:
            processor_func(item)