Compare commits

..
Author SHA1 Message Date
testclient-admin dc41e7e068 security-analysis/policy.yml: medium блокирует merge
security/scan 0 findings, 0 blocking
security / scan (pull_request) Successful in 52s
security/review 0 findings, 0 blocking
security / review (pull_request) Successful in 1m20s
security / deep-audit (pull_request) Skipped
2026-09-01 06:26:16 +00:00
3 changed files with 6 additions and 198 deletions
+1 -66
View File
@@ -1,6 +1,6 @@
# Модуль учёта заказов клиентов # Модуль учёта заказов клиентов
Простой HTTP-сервис для управления заказами и заметками клиентов. Простой HTTP-сервис для управления заказами клиентов.
## Структура ## Структура
@@ -16,14 +16,6 @@
- `user_id` — владелец заказа - `user_id` — владелец заказа
- `amount` — сумма заказа - `amount` — сумма заказа
### Хранение заметок
Заметки хранятся в памяти в словаре `notes`. Каждая заметка имеет:
- `id` — уникальный идентификатор
- `user_id` — владелец заметки
- `title` — заголовок заметки
- `content` — содержимое заметки
### Эндпоинты ### Эндпоинты
#### GET /orders/{id} #### GET /orders/{id}
@@ -65,63 +57,6 @@ curl -X POST -H "Authorization: Bearer <your-token>" \
Бэкап выполняется асинхронно через rsync. Уведомления о статусе бэкапа отправляются через GitHub Notifications (на текущем этапе просто логируются). Бэкап выполняется асинхронно через rsync. Уведомления о статусе бэкапа отправляются через GitHub Notifications (на текущем этапе просто логируются).
### Быстрый старт
#### Создание заметки
```bash
curl -X POST -H "Authorization: Bearer token_user123" \
-H "Content-Type: application/json" \
-d '{"title": "Моя заметка", "content": "Текст заметки"}' \
http://localhost:8000/notes
```
#### Получение заметки
```bash
curl -H "Authorization: Bearer token_user123" \
http://localhost:8000/notes/1
```
#### Получение списка заметок пользователя
```bash
curl -H "Authorization: Bearer token_user123" \
http://localhost:8000/notes
```
#### Обновление заметки
```bash
curl -X POST -H "Authorization: Bearer token_user123" \
-H "Content-Type: application/json" \
-d '{"title": "Новый заголовок", "content": "Новый текст"}' \
http://localhost:8000/notes/1
```
#### Удаление заметки
```bash
curl -X GET -H "Authorization: Bearer token_user123" \
http://localhost:8000/notes/1/delete
```
#### Экспорт заметки в файл
```bash
curl -X GET -H "Authorization: Bearer token_user123" \
"http://localhost:8000/notes/export?path=/tmp/note.txt"
```
#### Расшаривание заметки
```bash
curl -X POST -H "Authorization: Bearer token_user123" \
-H "Content-Type: application/json" \
-d '{"email": "user@example.com", "webhook_url": "https://webhook.site/abc123"}' \
http://localhost:8000/notes/1/share
```
## Запуск ## Запуск
```bash ```bash
-128
View File
@@ -12,7 +12,6 @@ import re
import subprocess import subprocess
orders = {} orders = {}
notes = {}
class OrdersHandler(BaseHTTPRequestHandler): class OrdersHandler(BaseHTTPRequestHandler):
@@ -65,55 +64,6 @@ class OrdersHandler(BaseHTTPRequestHandler):
self.send_json_response(200, orders[order_id]) self.send_json_response(200, orders[order_id])
else: else:
self.send_json_response(404, {"error": "Order not found"}) 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": elif self.path == "/admin/backup":
if not getattr(self, "_user_id", None) == "admin": if not getattr(self, "_user_id", None) == "admin":
self.send_json_response(403, {"error": "Forbidden - admin access required"}) self.send_json_response(403, {"error": "Forbidden - admin access required"})
@@ -142,65 +92,6 @@ class OrdersHandler(BaseHTTPRequestHandler):
self.send_json_response(200, orders[order_id]) self.send_json_response(200, orders[order_id])
else: else:
self.send_json_response(400, {"error": "Missing amount field"}) 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": elif self.path == "/admin/backup":
if not self.check_auth(): if not self.check_auth():
self.send_json_response(401, {"error": "Unauthorized"}) self.send_json_response(401, {"error": "Unauthorized"})
@@ -275,25 +166,6 @@ class OrdersHandler(BaseHTTPRequestHandler):
print(f"[GitHub Notification] {message}") print(f"[GitHub Notification] {message}")
pass 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): def log_message(self, format, *args):
print(f"[{self.log_date_time_string()}] {format % args}") print(f"[{self.log_date_time_string()}] {format % args}")
+5 -4
View File
@@ -1,7 +1,8 @@
# Какие находки блокируют merge. Читает security.yml на каждом
# скане; отсутствие файла — те же значения по умолчанию.
security: security:
block_on: block_on:
- critical - critical
- high - high
- medium
min_confidence: 0.7 min_confidence: 0.7
scan_enabled: true
review_enabled: true