Network Security Basics — Disable Certificate Verification

import hmac
import hashlib
import os

# ── INTEGRITY DEMO ──────────────────────────────────────────────────────────
# We use HMAC-SHA256 to create a Message Authentication Code (MAC).
# Both sender and receiver share a secret key.
# If even one byte of the message changes, the MAC will not match.

def create_mac(secret_key: bytes, message: bytes) -> str:
    """Produce a hex MAC for a message using a shared secret key."""
    mac = hmac.new(secret_key, message, hashlib.sha256)
    return mac.hexdigest()   # hex string is safe to transmit alongside the message

def verify_mac(secret_key: bytes, message: bytes, received_mac: str) -> bool:
    """Recompute the MAC and compare using constant-time comparison.
    
    hmac.compare_digest prevents timing attacks — an attacker can't
    tell 'how close' a forged MAC is by measuring response time.
    """
    expected_mac = create_mac(secret_key, message)
    return hmac.compare_digest(expected_mac, received_mac)


if __name__ == "__main__":
    shared_secret = os.urandom(32)   # 256-bit random key, never hardcoded in real code

    original_message = b"Transfer $500 to account 9982"
    mac_tag = create_mac(shared_secret, original_message)

    print("=== INTEGRITY CHECK DEMO ===")
    print(f"Original message : {original_message.decode()}")
    print(f"MAC tag          : {mac_tag[:16]}...  (truncated for display)")

    # Scenario 1: Message arrives unchanged
    is_valid = verify_mac(shared_secret, original_message, mac_tag)
    print(f"\nUnmodified message valid? {is_valid}")   # True

    # Scenario 2: Man-in-the-middle flips one digit in the amount
    tampered_message = b"Transfer $900 to account 9982"
    is_valid_after_tamper = verify_mac(shared_secret, tampered_message, mac_tag)
    print(f"Tampered  message valid? {is_valid_after_tamper}")   # False

    print("\n--- Result ---")
    if not is_valid_after_tamper:
        print("ALERT: Message integrity violated. Reject and log this event.")

import socket
import concurrent.futures
from typing import List, Tuple

# A minimal TCP port scanner — the same core logic used by tools like nmap.
# Understanding this helps you see EXACTLY what an attacker sees when they
# probe your server. Only scan hosts you own or have explicit permission to scan.

TARGET_HOST = "127.0.0.1"        # Scanning localhost — safe to run anywhere
PORTS_TO_CHECK = range(1, 1025)  # Well-known ports (IANA-reserved range)
CONNECT_TIMEOUT_SECONDS = 0.5    # Short timeout keeps scan fast

def probe_port(host: str, port: int) -> Tuple[int, bool, str]:
    """Attempt a TCP handshake. If it succeeds, the port is open.
    
    A completed TCP SYN-ACK exchange means a process is listening.
    A RST or timeout means the port is closed or filtered.
    """
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as tcp_socket:
            tcp_socket.settimeout(CONNECT_TIMEOUT_SECONDS)
            result_code = tcp_socket.connect_ex((host, port))  # 0 = success
            is_open = (result_code == 0)

            # Try to resolve the service name (e.g. 80 -> 'http')
            try:
                service_name = socket.getservbyport(port, "tcp")
            except OSError:
                service_name = "unknown"

            return port, is_open, service_name
    except (socket.timeout, ConnectionRefusedError, OSError):
        return port, False, "unreachable"

def scan_host(host: str, ports) -> List[Tuple[int, str]]:
    """Scan multiple ports in parallel using a thread pool."""
    open_ports = []
    # ThreadPoolExecutor lets us fire many simultaneous connection attempts
    with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
        futures = {executor.submit(probe_port, host, port): port for port in ports}
        for future in concurrent.futures.as_completed(futures):
            port, is_open, service = future.result()
            if is_open:
                open_ports.append((port, service))
    return sorted(open_ports)  # sort by port number for readability


if __name__ == "__main__":
    print(f"Scanning {TARGET_HOST} — ports 1-1024 ...\n")
    discovered = scan_host(TARGET_HOST, PORTS_TO_CHECK)

    if discovered:
        print(f"{'PORT':<8} {'SERVICE':<16} STATUS")
        print("-" * 35)
        for port_number, service_name in discovered:
            print(f"{port_number:<8} {service_name:<16} OPEN")
    else:
        print("No open ports found in range 1-1024.")

    print(f"\nTotal open ports: {len(discovered)}")
    print("Each open port is a potential entry point — close what you don't need.")

import ssl
import socket
import json

# This script makes a real TLS connection and inspects the certificate chain.
# It shows you exactly what your browser validates silently on every HTTPS request.

TARGET_HOST = "httpbin.org"   # A public test server — fine to query
HTTPS_PORT  = 443
HTTP_TIMEOUT_SECONDS = 5

def inspect_tls_connection(host: str, port: int) -> dict:
    """Open a verified TLS connection and extract security metadata."""

    # ssl.create_default_context() loads the OS certificate store.
    # It enforces hostname verification and certificate chain validation
    # automatically — this is the safe default, not the permissive one.
    tls_context = ssl.create_default_context()

    raw_socket = socket.create_connection((host, port), timeout=HTTP_TIMEOUT_SECONDS)

    # wrap_socket upgrades the plain TCP socket to TLS.
    # server_hostname is needed for SNI (Server Name Indication) so the
    # server knows which certificate to present when hosting multiple domains.
    tls_socket = tls_context.wrap_socket(raw_socket, server_hostname=host)

    # --- Extract metadata AFTER the handshake completes ---
    cipher_name, tls_version, key_bits = tls_socket.cipher()
    peer_cert = tls_socket.getpeercert()  # parsed certificate dict

    # Pull the Subject Common Name and issuer from the cert
    subject_fields  = dict(field for entry in peer_cert["subject"]  for field in entry)
    issuer_fields   = dict(field for entry in peer_cert["issuer"]   for field in entry)

    security_info = {
        "tls_version"       : tls_version,
        "cipher_suite"      : cipher_name,
        "cipher_key_bits"   : key_bits,
        "cert_common_name"  : subject_fields.get("commonName", "N/A"),
        "cert_issuer_org"   : issuer_fields.get("organizationName", "N/A"),
        "cert_valid_from"   : peer_cert.get("notBefore", "N/A"),
        "cert_valid_until"  : peer_cert.get("notAfter",  "N/A"),
        "hostname_verified" : True   # wrap_socket raised if verification failed
    }

    tls_socket.close()
    raw_socket.close()
    return security_info


if __name__ == "__main__":
    print(f"Establishing TLS connection to {TARGET_HOST}:{HTTPS_PORT}...\n")

    try:
        info = inspect_tls_connection(TARGET_HOST, HTTPS_PORT)
        print("=== TLS Handshake Security Summary ===")
        for label, value in info.items():
            print(f"  {label:<22}: {value}")

        # Warn if the server is using an outdated TLS version
        if info["tls_version"] in ("TLSv1", "TLSv1.1"):
            print("\n  WARNING: Server supports deprecated TLS version — upgrade required.")
        else:
            print("\n  TLS version is current and secure.")

    except ssl.SSLCertVerificationError as cert_error:
        # This fires if the cert is self-signed, expired, or hostname mismatches
        print(f"Certificate verification FAILED: {cert_error}")
        print("Do NOT ignore this error in production — it means you may be talking to an impersonator.")

import time
from collections import defaultdict, deque
from typing import Tuple

# A sliding-window rate limiter — the first line of defense against
# credential stuffing, brute-force login, and DDoS at the application layer.
#
# A fixed-window counter (e.g. '100 requests per minute') has a known flaw:
# an attacker can send 100 at 00:59 and 100 at 01:00 — 200 requests in 2 seconds.
# A sliding window fixes this by tracking actual request timestamps.

class SlidingWindowRateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        """
        max_requests  : how many requests are allowed per window
        window_seconds: the rolling time window in seconds
        """
        self.max_requests     = max_requests
        self.window_seconds   = window_seconds
        # Maps client_id -> deque of timestamps for requests in the current window
        self.request_history  = defaultdict(deque)

    def is_allowed(self, client_identifier: str) -> Tuple[bool, int]:
        """Check if a client is within their rate limit.
        
        Returns (allowed: bool, requests_remaining: int).
        The client_identifier could be an IP address, user ID, or API key.
        """
        current_time    = time.monotonic()            # monotonic avoids clock-skew bugs
        window_start    = current_time - self.window_seconds
        client_requests = self.request_history[client_identifier]

        # Evict timestamps older than the window — this is what makes it 'sliding'
        while client_requests and client_requests[0] < window_start:
            client_requests.popleft()

        requests_in_window = len(client_requests)

        if requests_in_window >= self.max_requests:
            # Deny — client has exhausted their quota
            return False, 0

        # Allow — record this request
        client_requests.append(current_time)
        remaining = self.max_requests - len(client_requests)
        return True, remaining


if __name__ == "__main__":
    # Simulate a login endpoint: max 5 attempts per 10-second window
    login_limiter = SlidingWindowRateLimiter(max_requests=5, window_seconds=10)

    attacker_ip = "192.168.1.105"  # Simulated credential-stuffing attacker
    legit_user  = "10.0.0.42"

    print("=== Credential Stuffing Defense Simulation ===")
    print(f"Limit: {login_limiter.max_requests} requests per {login_limiter.window_seconds}s\n")

    # Attacker fires 8 rapid login attempts
    for attempt_number in range(1, 9):
        allowed, remaining = login_limiter.is_allowed(attacker_ip)
        status = "ALLOWED  " if allowed else "BLOCKED  "
        print(f"Attacker attempt #{attempt_number}: {status} | Remaining quota: {remaining}")

    print()
    # Legitimate user makes one request — their window is clean
    allowed, remaining = login_limiter.is_allowed(legit_user)
    print(f"Legit user request:  {'ALLOWED' if allowed else 'BLOCKED'} | Remaining quota: {remaining}")

#!/bin/bash
# Example iptables rules for a three-tier segmentation on a single host
# This is a simplified demonstration; in production use cloud security groups or nftables.

# ── Variables ────────────────────────────────────────────────────────────
WEB_SUBNET="10.0.1.0/24"
APP_SUBNET="10.0.2.0/24"
DB_SUBNET="10.0.3.0/24"

# ── Flush existing rules (careful: will break existing connections) ──────
iptables -F
iptables -X

# ── Default policies ──────────────────────────────────────────────────────
iptables -P INPUT DROP
iptables -P FORWARD DROP
iptables -P OUTPUT ACCEPT

# ── Allow established connections ─────────────────────────────────────────
iptables -A INPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT

# ── Web tier: allow HTTP/HTTPS from anywhere ──────────────────────────────
iptables -A INPUT -p tcp --dport 80 -j ACCEPT
iptables -A INPUT -p tcp --dport 443 -j ACCEPT

# ── App tier: allow traffic only from web subnet to app ports ─────────────
iptables -A INPUT -s $WEB_SUBNET -p tcp --dport 8080 -j ACCEPT

# ── Database tier: allow traffic only from app subnet to DB port ───────────
iptables -A INPUT -s $APP_SUBNET -p tcp --dport 5432 -j ACCEPT

# ── SSH only from management IP ───────────────────────────────────────────
iptables -A INPUT -s 203.0.113.0/24 -p tcp --dport 22 -j ACCEPT

# ── Log dropped packets (optional) ───────────────────────────────────────
iptables -A INPUT -j LOG --log-prefix "FW-DROP: "

echo "Rules applied. Use iptables -L -v to see them."

// io.thecodeforge — cs-fundamentals tutorial

import pyshark
from collections import defaultdict
import time

# Production baseline: internal DNS server IP
INTERNAL_DNS = "10.0.20.53"
DOMAIN_WHITELIST = {"thecodeforge.io", "apt-update.corp", "vpn.corp"}

def monitor_dns_anomalies(interface="eth0", duration_sec=60):
    suspicious_ips = defaultdict(set)
    capture = pyshark.LiveCapture(interface=interface, bpf_filter="udp port 53")

    print(f"[*] Monitoring DNS on {interface} for {duration_sec}s...")
    start = time.time()
    
    for packet in capture.sniff_continuously():
        if time.time() - start > duration_sec:
            break
        try:
            src_ip = packet.ip.src
            qname = packet.dns.qry_name
            if src_ip != INTERNAL_DNS and qname not in DOMAIN_WHITELIST:
                suspicious_ips[src_ip].add(qname)
        except AttributeError:
            continue

    for ip, domains in suspicious_ips.items():
        if len(domains) > 10:
            print(f"[!] ALERT: {ip} resolved {len(domains)} unknown domains")
            for d in list(domains)[:5]:
                print(f"    -> {d}")

// io.thecodeforge — cs-fundamentals tutorial

import ssl
import socket
import datetime

CRITICAL_SERVICES = [
    ("api.payments.corp", 443),
    ("auth.sso.corp", 443),
    ("mail.route.corp", 25),
]

def check_cert_expiry(hostname, port):
    context = ssl.create_default_context()
    with socket.create_connection((hostname, port), timeout=5) as sock:
        with context.wrap_socket(sock, server_hostname=hostname) as ssock:
            cert = ssock.getpeercert()
            expiry = datetime.datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
            remaining = (expiry - datetime.datetime.utcnow()).days
            return hostname, remaining

print("Service                Days to Expiry")
print("-" * 40)
for host, port in CRITICAL_SERVICES:
    try:
        _, days = check_cert_expiry(host, port)
        status = "EXPIRED" if days < 0 else f"{days:3d}"
        print(f"{host:22s} {status}")
    except Exception as e:
        print(f"{host:22s} FAIL/{type(e).__name__}")