294 lines
12 KiB
Python
294 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/"):
|
|
self._handle_notes_get()
|
|
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.startswith("/notes/"):
|
|
self._handle_notes_post()
|
|
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 _handle_notes_get(self):
|
|
if not self.check_auth():
|
|
self.send_json_response(401, {"error": "Unauthorized"})
|
|
return
|
|
path_parts = self.path.split("/")
|
|
if len(path_parts) == 3 and path_parts[2]:
|
|
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 len(path_parts) == 3 and path_parts[1] == "export" and "path=" in self.path:
|
|
self._handle_notes_export()
|
|
else:
|
|
self.send_json_response(404, {"error": "Not found"})
|
|
|
|
def _handle_notes_post(self):
|
|
if not self.check_auth():
|
|
self.send_json_response(401, {"error": "Unauthorized"})
|
|
return
|
|
path_parts = self.path.split("/")
|
|
if len(path_parts) == 2:
|
|
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"],
|
|
"created_at": time.time()
|
|
}
|
|
self.send_json_response(201, notes[note_id])
|
|
else:
|
|
self.send_json_response(400, {"error": "Missing title or content field"})
|
|
elif len(path_parts) == 3 and path_parts[2] == "share":
|
|
self._handle_notes_share(path_parts[1])
|
|
else:
|
|
self.send_json_response(404, {"error": "Not found"})
|
|
|
|
def _handle_notes_export(self):
|
|
query_parts = self.path.split("?")
|
|
if len(query_parts) != 2:
|
|
self.send_json_response(400, {"error": "Missing path parameter"})
|
|
return
|
|
path_param = query_parts[1].replace("path=", "")
|
|
if not path_param:
|
|
self.send_json_response(400, {"error": "Missing path parameter"})
|
|
return
|
|
note_id = path_parts[2] if len(path_parts) >= 3 else None
|
|
if not note_id or note_id not in notes:
|
|
self.send_json_response(404, {"error": "Note not found"})
|
|
return
|
|
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
|
self.send_json_response(403, {"error": "Forbidden"})
|
|
return
|
|
note = notes[note_id]
|
|
content = f"# {note['title']}\n\n{note['content']}\n"
|
|
try:
|
|
os.makedirs(os.path.dirname(path_param), exist_ok=True)
|
|
with open(path_param, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
self.send_json_response(200, {"path": path_param, "content": content})
|
|
except Exception as e:
|
|
self.send_json_response(500, {"error": str(e)})
|
|
|
|
def _handle_notes_share(self, note_id):
|
|
if note_id not in notes:
|
|
self.send_json_response(404, {"error": "Note not found"})
|
|
return
|
|
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, "note_id": note_id})
|
|
else:
|
|
self.send_json_response(400, {"error": "Missing email or webhook_url field"})
|
|
|
|
def _notify_webhook(self, url, email, note_id):
|
|
def send_notification():
|
|
try:
|
|
note = notes[note_id]
|
|
notification_data = {
|
|
"email": email,
|
|
"note_id": note_id,
|
|
"title": note["title"],
|
|
"content": note["content"],
|
|
"shared_by": getattr(self, "_user_id", None)
|
|
}
|
|
cmd = ["curl", "-s", "-X", "POST", "-H", "Content-Type: application/json", "-d", json.dumps(notification_data), url]
|
|
subprocess.run(cmd, capture_output=True, text=True, timeout=10)
|
|
print(f"Webhook notification sent to {url} for note {note_id}")
|
|
except Exception as e:
|
|
print(f"Failed to send webhook notification: {e}")
|
|
thread = threading.Thread(target=send_notification)
|
|
thread.start()
|
|
|
|
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 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()
|