271 lines
12 KiB
Python
271 lines
12 KiB
Python
#!/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.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 == "/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.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
|
|
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()
|