test.py

#!/usr/bin/env python3
import threading
import requests
import hashlib
import os
import json

# Adjust the HTTP server base URL if necessary
SERVER = "http://127.0.0.1:8080"
BASE_DIR = os.path.dirname(__file__)
RESOURCE_DIR = os.path.join(BASE_DIR, 'resources')

# List of files to test and verify, update according to your folder structure
files = ['logo.png', 'photo.jpg']  # .png and .jpg match your visible files

def file_checksum(path):
    with open(path, 'rb') as f:
        return hashlib.sha256(f.read()).hexdigest()

original_checksums = {f: file_checksum(os.path.join(RESOURCE_DIR, f)) for f in files}

def download_file(file):
    url = f"{SERVER}/{file}"
    headers = {'Host': '127.0.0.1:8080'}
    try:
        r = requests.get(url, headers=headers)
        status = r.status_code
        content_len = len(r.content)
        checksum = hashlib.sha256(r.content).hexdigest()
        match = "✓" if checksum == original_checksums[file] else "✗"
        print(f"{file}: {status}, {content_len} bytes, checksum match: {match}")
        return status == 200
    except Exception as e:
        print(f"{file}: ERROR - {str(e)}")
        return False

def test_request(path, expected_status, description):
    try:
        headers = {'Host': '127.0.0.1:8080'}
        r = requests.get(f"{SERVER}{path}", headers=headers)
        success = r.status_code == expected_status
        status_text = "✅" if success else "❌"
        print(f"{description}: {status_text} (got {r.status_code}, expected {expected_status})")
        return success
    except Exception as e:
        print(f"{description}: ❌ (ERROR - {str(e)})")
        return False

print("\n=== Basic Functionality Tests ===")
basic_tests = [
    ('/', 200, 'GET / → index.html'),
    ('/about.html', 200, 'GET /about.html'),
    ('/logo.png', 200, 'GET /logo.png'),
    ('/photo.jpg', 200, 'GET /photo.jpg'),
    ('/sample.txt', 200, 'GET /sample.txt'),
    ('/nonexistent.png', 404, 'GET /nonexistent.png → 404'),
]

basic_results = []
for path, expected, desc in basic_tests:
    result = test_request(path, expected, desc)
    basic_results.append(result)

# POST upload test
try:
    headers = {'Host': '127.0.0.1:8080', 'Content-Type': 'application/json'}
    data = json.dumps({"test": "data", "timestamp": "2025-10-09"})
    r = requests.post(f"{SERVER}/upload", headers=headers, data=data)
    upload_success = r.status_code == 201
    print(f"POST /upload with JSON: {'✅' if upload_success else '❌'} (got {r.status_code})")
    basic_results.append(upload_success)
except Exception as e:
    print(f"POST /upload with JSON: ❌ (ERROR - {str(e)})")
    basic_results.append(False)

# PUT test (should be 405)
try:
    headers = {'Host': '127.0.0.1:8080'}
    r = requests.put(f"{SERVER}/index.html", headers=headers)
    put_success = r.status_code == 405
    print(f"PUT /index.html → 405: {'✅' if put_success else '❌'} (got {r.status_code})")
    basic_results.append(put_success)
except Exception as e:
    print(f"PUT /index.html → 405: ❌ (ERROR - {str(e)})")
    basic_results.append(False)

# POST with non-JSON (should be 415)
try:
    headers = {'Host': '127.0.0.1:8080', 'Content-Type': 'text/plain'}
    r = requests.post(f"{SERVER}/upload", headers=headers, data="not json")
    json_success = r.status_code == 415
    print(f"POST /upload with non-JSON → 415: {'✅' if json_success else '❌'} (got {r.status_code})")
    basic_results.append(json_success)
except Exception as e:
    print(f"POST /upload with non-JSON → 415: ❌ (ERROR - {str(e)})")
    basic_results.append(False)

print("\n=== Security Tests ===")
security_tests = [
    ('/../etc/passwd', 403, 'GET /../etc/passwd → 403'),
    ('/./././../config', 403, 'GET /./././../config → 403'),
    ('/../../../etc/passwd', 403, 'GET /../../../etc/passwd → 403'),
    ('/uploads/../../server.py', 403, 'GET /uploads/../../server.py → 403'),
]

security_results = []
for path, expected, desc in security_tests:
    result = test_request(path, expected, desc)
    security_results.append(result)

# Host header tests
try:
    headers = {'Host': 'evil.com:8080'}
    r = requests.get(f"{SERVER}/", headers=headers)
    host_success = r.status_code == 403
    print(f"Request with Host: evil.com → 403: {'✅' if host_success else '❌'} (got {r.status_code})")
    security_results.append(host_success)
except Exception as e:
    print(f"Request with Host: evil.com → 403: ❌ (ERROR - {str(e)})")
    security_results.append(False)

print("\n=== Concurrency Tests ===")
results = {}
threads = []
for f in files:
    t = threading.Thread(target=lambda file=f: results.update({file: download_file(file)}))
    t.start()
    threads.append(t)
for t in threads:
    t.join()

concurrency_results = []
for f in files:
    success = results.get(f, False)
    print(f"Download {f}: {'✅' if success else '❌'}")
    concurrency_results.append(success)

print("\nQueue / Thread Pool Saturation Test:")
threads = []
for i in range(3):  # 3 rounds
    for f in files:
        t = threading.Thread(target=download_file, args=(f,))
        t.start()
        threads.append(t)
for t in threads:
    t.join()
print("All queue/overload downloads attempted")

print("\n" + "="*50)
print("FINAL TEST RESULTS")
print("="*50)
basic_passed = sum(basic_results)
security_passed = sum(security_results)
concurrency_passed = sum(concurrency_results)

print(f"Basic Functionality: {basic_passed}/{len(basic_results)} tests passed")
print(f"Security Tests: {security_passed}/{len(security_results)} tests passed")
print(f"Concurrency Tests: {concurrency_passed}/{len(concurrency_results)} tests passed")

if security_passed == len(security_results):
    print("🎉 ALL SECURITY TESTS PASSED! Path traversal is properly blocked.")
else:
    print("⚠️  SECURITY ISSUES REMAIN: Some path traversal attempts are not blocked.")

print("="*50)