#!/usr/bin/env python3 """Модуль учёта заказов клиентов""" from http.server import HTTPServer, BaseHTTPRequestHandler import json import threading import time import hashlib import hmac import os import re import subprocess orders = {} notes = {} class OrdersHandler(BaseHTTPRequestHandler): def check_auth(self): auth_header = self.headers.get("Authorization", "") if not auth_header.startswith("Bearer "): return False token = auth_header[7:] return self._validate_auth() def _validate_token(self, token): if not token.startswith("token_"): return False parts = token.split(":") if len(parts) >= 2: self._user_id = parts[1] return True self._user_id = None return False def _validate_auth(self): auth_header = self.headers.get("Authorization", "") if not auth_header.startswith("Bearer "): return False token = auth_header[7:] if self._validate_token(token): return True admin_token = os.getenv("ADMIN_TOKEN", "") if admin_token and token == admin_token: self._user_id = "admin" return True return False def send_json_response(self, status_code, data): self.send_response(status_code) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(data).encode()) def do_GET(self): if self.path.startswith("/orders/"): if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return order_id = self.path.split("/")[-1] if order_id in orders: if orders[order_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return self.send_json_response(200, orders[order_id]) else: self.send_json_response(404, {"error": "Order not found"}) elif self.path == "/notes": if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return user_notes = [n for n in notes.values() if n.get("user_id") == getattr(self, "_user_id", None)] self.send_json_response(200, user_notes) elif self.path.startswith("/notes/"): if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return path_parts = self.path.split("/") if len(path_parts) == 3: note_id = path_parts[2] if note_id in notes: if notes[note_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return self.send_json_response(200, notes[note_id]) else: self.send_json_response(404, {"error": "Note not found"}) elif self.path.startswith("/notes/export"): query_params = self.path.split("?") if len(query_params) > 1: params = dict(p.split("=") for p in query_params[1].split("&")) if "path" in params: note_id = params["path"] if note_id in notes: if notes[note_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return note_content = notes[note_id] self.send_json_response(200, {"content": note_content}) else: self.send_json_response(404, {"error": "Note not found"}) else: self.send_json_response(400, {"error": "Missing path parameter"}) else: self.send_json_response(400, {"error": "Missing path parameter"}) elif self.path.endswith("/delete"): note_id = self.path.split("/")[2] if note_id in notes: if notes[note_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return del notes[note_id] self.send_json_response(200, {"status": "deleted"}) else: self.send_json_response(404, {"error": "Note not found"}) return elif self.path == "/admin/backup": if not getattr(self, "_user_id", None) == "admin": self.send_json_response(403, {"error": "Forbidden - admin access required"}) return self.send_json_response(200, {"status": "backup_endpoint_ready"}) else: self.send_json_response(404, {"error": "Not found"}) def do_POST(self): if self.path.startswith("/orders/"): if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return order_id = self.path.split("/")[-1] content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode() data = json.loads(body) if "amount" in data: if order_id not in orders: orders[order_id] = {"id": order_id, "user_id": getattr(self, "_user_id", None), "amount": 0} else: if orders[order_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return orders[order_id]["amount"] = data["amount"] self.send_json_response(200, orders[order_id]) else: self.send_json_response(400, {"error": "Missing amount field"}) elif self.path == "/notes": if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode() data = json.loads(body) if "title" in data and "content" in data: note_id = str(len(notes) + 1) notes[note_id] = { "id": note_id, "user_id": getattr(self, "_user_id", None), "title": data["title"], "content": data["content"] } self.send_json_response(201, notes[note_id]) else: self.send_json_response(400, {"error": "Missing title or content field"}) elif self.path.startswith("/notes/") and self.path.count("/") == 2: note_id = self.path.split("/")[-1] content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode() data = json.loads(body) if note_id in notes: if notes[note_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return if "title" in data: notes[note_id]["title"] = data["title"] if "content" in data: notes[note_id]["content"] = data["content"] self.send_json_response(200, notes[note_id]) else: self.send_json_response(404, {"error": "Note not found"}) elif self.path.endswith("/share"): if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return path_parts = self.path.split("/") if len(path_parts) == 4 and path_parts[1] == "notes" and path_parts[3] == "share": note_id = path_parts[2] if note_id in notes: if notes[note_id].get("user_id") != getattr(self, "_user_id", None): self.send_json_response(403, {"error": "Forbidden"}) return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode() data = json.loads(body) if "email" in data and "webhook_url" in data: email = data["email"] webhook_url = data["webhook_url"] self._notify_webhook(webhook_url, email, note_id) self.send_json_response(200, {"status": "shared", "email": email}) else: self.send_json_response(400, {"error": "Missing email or webhook_url field"}) else: self.send_json_response(404, {"error": "Note not found"}) else: self.send_json_response(404, {"error": "Not found"}) elif self.path == "/admin/backup": if not self.check_auth(): self.send_json_response(401, {"error": "Unauthorized"}) return if not getattr(self, "_user_id", None) == "admin": self.send_json_response(403, {"error": "Forbidden - admin access required"}) return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode() data = json.loads(body) if "host" in data: backup_host = data["host"] self._perform_backup(backup_host) self.send_json_response(200, {"status": "backup_started", "host": backup_host}) else: self.send_json_response(400, {"error": "Missing host field"}) else: self.send_json_response(404, {"error": "Not found"}) def _perform_backup(self, host): def run_backup(): try: validated_host = self._validate_host(host) if not validated_host: print(f"Invalid host: {host}") self._notify_github(f"Backup failed: invalid host {host}") return cmd = ["rsync", "-avz", "/workspace/", f"{validated_host}:/backup/orders/"] result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if result.returncode == 0: print(f"Backup to {validated_host} completed successfully") self._notify_github(f"Backup to {validated_host} completed successfully") else: print(f"Backup failed: {result.stderr}") self._notify_github(f"Backup to {validated_host} failed: {result.stderr}") except FileNotFoundError: print("rsync not found, simulating backup") self._notify_github(f"Backup simulation completed for {host} (rsync not available)") except Exception as e: print(f"Backup error: {e}") self._notify_github(f"Backup to {host} error: {str(e)}") thread = threading.Thread(target=run_backup) thread.start() def _validate_host(self, host): if not host: return None host = host.strip() if len(host) > 255: return None if host.startswith('--'): return None if ' ' in host: return None ip_pattern = r'^(\d{1,3}\.){3}\d{1,3}$' domain_pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$' if re.match(ip_pattern, host): parts = host.split('.') if all(0 <= int(p) <= 255 for p in parts): return host return None if re.match(domain_pattern, host): return host return None def _notify_github(self, message): token = os.getenv("GITHUB_TOKEN") if token: print(f"[GitHub Notification] {message}") else: print(f"[GitHub Notification] {message}") pass def _notify_webhook(self, webhook_url, email, note_id): def send_notification(): try: import urllib.request import json import urllib.parse parsed = urllib.parse.urlparse(webhook_url) if parsed.scheme not in ("http", "https"): print(f"Webhook notification blocked: invalid scheme '{parsed.scheme}'") return data = json.dumps({"email": email, "note_id": note_id}).encode() req = urllib.request.Request(webhook_url, data=data, headers={"Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=10) print(f"Webhook notification sent to {webhook_url} for email {email}") except Exception as e: print(f"Webhook notification failed: {e}") thread = threading.Thread(target=send_notification) thread.start() def log_message(self, format, *args): print(f"[{self.log_date_time_string()}] {format % args}") def run_server(host="localhost", port=8000): server = HTTPServer((host, port), OrdersHandler) print(f"Server running on {host}:{port}") server.serve_forever() if __name__ == "__main__": run_server()