384 lines
14 KiB
Python
384 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
HTTPS File Transfer Server for OpenWRT/Notebook
|
|
Secure file upload, download, and management over HTTPS
|
|
"""
|
|
|
|
import ssl
|
|
import os
|
|
import json
|
|
import mimetypes
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from datetime import datetime
|
|
import logging
|
|
|
|
# ============================================================================
|
|
# Configuration
|
|
# ============================================================================
|
|
|
|
# Server settings
|
|
HOST = "0.0.0.0"
|
|
HTTP_PORT = 8080
|
|
HTTPS_PORT = 4443
|
|
|
|
# File settings
|
|
UPLOAD_DIR = "./uploads"
|
|
CERT_DIR = "./ssl-dir"
|
|
SERVER_CERT = os.path.join(CERT_DIR, "server.crt")
|
|
SERVER_KEY = os.path.join(CERT_DIR, "server.key")
|
|
|
|
# Logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('server.log'),
|
|
logging.StreamHandler()
|
|
]
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ============================================================================
|
|
# Utility Functions
|
|
# ============================================================================
|
|
|
|
def ensure_upload_dir():
|
|
"""Create upload directory if it doesn't exist"""
|
|
Path(UPLOAD_DIR).mkdir(parents=True, exist_ok=True)
|
|
logger.info(f"Upload directory: {os.path.abspath(UPLOAD_DIR)}")
|
|
|
|
def ensure_ssl_certs():
|
|
"""Check if SSL certificates exist"""
|
|
if not os.path.exists(SERVER_CERT) or not os.path.exists(SERVER_KEY):
|
|
logger.error(f"SSL certificates not found in {CERT_DIR}")
|
|
logger.info("Generate certificates with:")
|
|
logger.info(f" mkdir -p {CERT_DIR}")
|
|
logger.info(f" openssl req -x509 -newkey rsa:2048 -nodes -days 3650 \\")
|
|
logger.info(f" -keyout {SERVER_KEY} -out {SERVER_CERT} \\")
|
|
logger.info(f" -subj '/C=US/ST=State/L=City/O=Org/CN=localhost'")
|
|
return False
|
|
return True
|
|
|
|
def get_file_list():
|
|
"""Get list of files in upload directory
|
|
Skips directories"""
|
|
try:
|
|
files = []
|
|
for filename in os.listdir(UPLOAD_DIR):
|
|
filepath = os.path.join(UPLOAD_DIR, filename)
|
|
if os.path.isfile(filepath):
|
|
size = os.path.getsize(filepath)
|
|
mtime = os.path.getmtime(filepath)
|
|
files.append({
|
|
"name": filename,
|
|
"size": size,
|
|
"modified": datetime.fromtimestamp(mtime).isoformat(),
|
|
"size_human": format_size(size)
|
|
})
|
|
return sorted(files, key=lambda x: x['modified'], reverse=True)
|
|
except Exception as e:
|
|
logger.error(f"Error listing files: {e}")
|
|
return []
|
|
|
|
def format_size(bytes_size):
|
|
"""Convert bytes to human readable format"""
|
|
for unit in ['B', 'KB', 'MB', 'GB']:
|
|
if bytes_size < 1024:
|
|
return f"{bytes_size:.1f}{unit}"
|
|
bytes_size /= 1024
|
|
return f"{bytes_size:.1f}TB"
|
|
|
|
def get_content_type(filename):
|
|
"""Get MIME type for file"""
|
|
mime_type, _ = mimetypes.guess_type(filename)
|
|
return mime_type or 'application/octet-stream'
|
|
|
|
# ============================================================================
|
|
# HTML/CSS/JavaScript
|
|
# ============================================================================
|
|
|
|
with open("index.html", "r", encoding="utf-8") as f:
|
|
HTML_TEMPLATE = f.read()
|
|
# ============================================================================
|
|
# Request Handler
|
|
# ============================================================================
|
|
|
|
class FileTransferHandler(BaseHTTPRequestHandler):
|
|
"""HTTP request handler for file transfer operations"""
|
|
|
|
def log_message(self, format, *args):
|
|
"""Override logging to use our logger"""
|
|
logger.info(f"{self.client_address[0]} - {format % args}")
|
|
|
|
def do_GET(self):
|
|
"""Handle GET requests"""
|
|
try:
|
|
# Parse URL
|
|
parsed_url = urllib.parse.urlparse(self.path)
|
|
path = parsed_url.path
|
|
query_params = urllib.parse.parse_qs(parsed_url.query)
|
|
|
|
# Route handling
|
|
if path == "/":
|
|
self.serve_index()
|
|
elif path == "/api/files":
|
|
self.serve_file_list()
|
|
elif path == "/download":
|
|
filename = query_params.get('file', [None])[0]
|
|
if filename:
|
|
self.serve_download(filename)
|
|
else:
|
|
self.send_error(400, "Missing file parameter")
|
|
elif path == "/status":
|
|
self.serve_status()
|
|
else:
|
|
self.send_error(404, "Not found")
|
|
|
|
except Exception as e:
|
|
logger.error(f"GET error: {e}")
|
|
self.send_error(500, "Internal server error")
|
|
|
|
def do_POST(self):
|
|
"""Handle POST requests"""
|
|
try:
|
|
parsed_url = urllib.parse.urlparse(self.path)
|
|
path = parsed_url.path
|
|
query_params = urllib.parse.parse_qs(parsed_url.path.split('?')[0] if '?' in self.path else '')
|
|
if path == "/upload":
|
|
self.handle_upload()
|
|
elif path == "/delete":
|
|
query_params = urllib.parse.parse_qs(self.path.split('?')[1] if '?' in self.path else '')
|
|
|
|
filename = query_params.get('file', [None])[0]
|
|
if filename:
|
|
self.handle_delete(filename)
|
|
else:
|
|
self.send_error(400, "Missing file parameter")
|
|
else:
|
|
self.send_error(404, "Not found")
|
|
|
|
except Exception as e:
|
|
logger.error(f"POST error: {e}")
|
|
self.send_error(500, "Internal server error")
|
|
|
|
def serve_index(self):
|
|
"""Serve the main HTML page"""
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "text/html; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(HTML_TEMPLATE.encode('utf-8'))
|
|
|
|
def serve_file_list(self):
|
|
"""Serve JSON list of files"""
|
|
files = get_file_list()
|
|
response = {
|
|
"files": files,
|
|
"total_files": len(files),
|
|
"upload_dir": os.path.abspath(UPLOAD_DIR)
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(response).encode('utf-8'))
|
|
|
|
def serve_download(self, filename):
|
|
"""Serve file download"""
|
|
# Security: prevent directory traversal
|
|
if ".." in filename or filename.startswith("/"):
|
|
self.send_error(403, "Forbidden")
|
|
return
|
|
|
|
filepath = os.path.join(UPLOAD_DIR, filename)
|
|
|
|
if not os.path.exists(filepath) or not os.path.isfile(filepath):
|
|
self.send_error(404, "File not found")
|
|
return
|
|
|
|
try:
|
|
with open(filepath, 'rb') as f:
|
|
file_size = os.path.getsize(filepath)
|
|
self.send_response(200)
|
|
self.send_header("Content-type", get_content_type(filename))
|
|
self.send_header("Content-Disposition", f'attachment; filename="{filename}"')
|
|
self.send_header("Content-Length", str(file_size))
|
|
self.end_headers()
|
|
|
|
# Send file in chunks
|
|
chunk_size = 8192
|
|
while True:
|
|
chunk = f.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
self.wfile.write(chunk)
|
|
|
|
logger.info(f"Downloaded: {filename}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Download error: {e}")
|
|
self.send_error(500, "Error downloading file")
|
|
|
|
def serve_status(self):
|
|
"""Serve server status"""
|
|
files = get_file_list()
|
|
total_size = sum(f['size'] for f in files)
|
|
|
|
status = {
|
|
"status": "online",
|
|
"timestamp": datetime.now().isoformat(),
|
|
"total_files": len(files),
|
|
"total_size": total_size,
|
|
"upload_dir": UPLOAD_DIR,
|
|
"https_port": HTTPS_PORT
|
|
}
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(status).encode()) #encode('utf-8')
|
|
|
|
def handle_upload(self):
|
|
"""Handle file upload"""
|
|
try:
|
|
content_length = int(self.headers.get('Content-Length', 0))
|
|
|
|
if content_length > 0:
|
|
# Parse multipart form data
|
|
content_type = self.headers.get('Content-Type', '')
|
|
|
|
if 'multipart/form-data' in content_type:
|
|
# Extract boundary
|
|
boundary = content_type.split("boundary=")[1].encode()
|
|
|
|
# Read data
|
|
body = self.rfile.read(content_length)
|
|
|
|
# Simple multipart parser
|
|
parts = body.split(b'--' + boundary)
|
|
for part in parts:
|
|
if b'filename=' in part:
|
|
# Extract filename
|
|
try:
|
|
filename_start = part.find(b'filename="') + 10
|
|
filename_end = part.find(b'"', filename_start)
|
|
filename = part[filename_start:filename_end].decode()
|
|
|
|
# Extract file data
|
|
data_start = part.find(b'\r\n\r\n') + 4
|
|
data_end = part.rfind(b'\r\n')
|
|
file_data = part[data_start:data_end]
|
|
|
|
# Save file
|
|
filepath = os.path.join(UPLOAD_DIR, filename)
|
|
with open(filepath, 'wb') as f:
|
|
f.write(file_data)
|
|
|
|
logger.info(f"Uploaded: {filename} ({len(file_data)} bytes)")
|
|
except Exception as e:
|
|
logger.error(f"Upload parse error: {e}")
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({"status": "success"}).encode())
|
|
else:
|
|
self.send_error(400, "No file provided")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Upload error: {e}")
|
|
self.send_error(500, "Upload failed")
|
|
|
|
def handle_delete(self, filename):
|
|
"""Handle file deletion"""
|
|
# Security: prevent directory traversal
|
|
if ".." in filename or filename.startswith("/"):
|
|
self.send_error(403, "Forbidden")
|
|
return
|
|
|
|
filepath = os.path.join(UPLOAD_DIR, filename)
|
|
|
|
if not os.path.exists(filepath):
|
|
self.send_error(404, "File not found")
|
|
return
|
|
|
|
try:
|
|
os.remove(filepath)
|
|
logger.info(f"Deleted: {filename}")
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-type", "application/json")
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({"status": "success"}).encode())
|
|
|
|
except Exception as e:
|
|
logger.error(f"Delete error: {e}")
|
|
self.send_error(500, "Delete failed")
|
|
|
|
# ============================================================================
|
|
# Server Initialization
|
|
# ============================================================================
|
|
|
|
def create_https_server():
|
|
"""Create and configure HTTPS server"""
|
|
server = HTTPServer((HOST, HTTPS_PORT), FileTransferHandler)
|
|
|
|
# Create SSL context
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
context.load_cert_chain(certfile=SERVER_CERT, keyfile=SERVER_KEY)
|
|
|
|
# Optional: Improve security
|
|
context.options |= ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3 | ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1
|
|
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
|
|
# Wrap socket
|
|
server.socket = context.wrap_socket(server.socket, server_side=True)
|
|
|
|
return server
|
|
|
|
def create_http_server():
|
|
"""Create plain HTTP server (for redirects if needed)"""
|
|
server = HTTPServer((HOST, HTTP_PORT), FileTransferHandler)
|
|
return server
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
print("=" * 70)
|
|
print(" HTTPS File Transfer Server")
|
|
print("=" * 70)
|
|
|
|
# Check requirements
|
|
if not ensure_ssl_certs():
|
|
print("\n❌ SSL certificates not found. Please generate them first.")
|
|
return
|
|
|
|
ensure_upload_dir()
|
|
|
|
# Create servers
|
|
https_server = create_https_server()
|
|
|
|
logger.info("=" * 70)
|
|
logger.info("File Transfer Server Starting")
|
|
logger.info("=" * 70)
|
|
logger.info(f"📁 Upload directory: {os.path.abspath(UPLOAD_DIR)}")
|
|
logger.info(f"🔒 HTTPS port: {HTTPS_PORT}")
|
|
logger.info(f"📝 Log file: server.log")
|
|
logger.info("=" * 70)
|
|
|
|
print(f"\n✅ Server starting...")
|
|
print(f"📱 Access from iPhone: https://localhost:{HTTPS_PORT}")
|
|
print(f"🌐 Access from network: https://<router-ip>:{HTTPS_PORT}")
|
|
print(f"\n⚠️ Self-signed certificate - tap 'Continue' on first visit")
|
|
print(f"📁 Files uploaded to: {os.path.abspath(UPLOAD_DIR)}")
|
|
print(f"📝 Logs: server.log\n")
|
|
|
|
try:
|
|
https_server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
logger.info("Server shutting down...")
|
|
https_server.shutdown()
|
|
print("\n✅ Server stopped")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|