142 lines
5.6 KiB
Python
Executable File
142 lines
5.6 KiB
Python
Executable File
#!/usr/bin/python3
|
|
from flask import Flask, make_response, request
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
app = Flask(__name__)
|
|
database_node_port = 5002
|
|
basic_path = "./files/db"
|
|
|
|
IMG_MIN, IMG_MAX = 1, 16
|
|
# Поля профиля, которые вообще могут существовать
|
|
ALLOWED_FIELDS = {"password", "first_name", "second_name", "email", "msg"}
|
|
|
|
# PATCH V4: database без auth. Даже если :38002 снова опубликуют на 0.0.0.0,
|
|
# без этого заголовка (его знают только logic/content) чужие запросы отсекаются.
|
|
INTERNAL = os.environ.get("RUSGRAM_INTERNAL", "e9b2f6d14a8c7035e1d0a6b8c4f2e7a9")
|
|
|
|
|
|
def require_internal():
|
|
if request.headers.get("X-Internal") != INTERNAL:
|
|
return make_response({"status": "forbidden"}, 403)
|
|
return None
|
|
|
|
|
|
def safe_username(username):
|
|
"""Имя юзера становится именем файла в db/users/ — режем всё, что уводит
|
|
запись/чтение за пределы каталога."""
|
|
if not username or len(username) > 64:
|
|
return False
|
|
if "/" in username or "\\" in username or username in (".", ".."):
|
|
return False
|
|
if "\x00" in username or any(ord(c) < 0x20 for c in username):
|
|
return False
|
|
return True
|
|
|
|
|
|
def user_path(username):
|
|
return os.path.join(basic_path, "users", username)
|
|
|
|
|
|
@app.route('/users/<string:username>/<string:field>', methods=["GET", "POST"])
|
|
def users(username, field):
|
|
denied = require_internal()
|
|
if denied is not None:
|
|
return denied
|
|
if not safe_username(username) or field not in ALLOWED_FIELDS:
|
|
return make_response({"status": "bad_request"}, 400)
|
|
path = user_path(username)
|
|
|
|
if request.method == 'GET':
|
|
if not os.path.exists(path):
|
|
return make_response({"status": "user_not_found"}, 404)
|
|
try:
|
|
with open(path, "r") as f:
|
|
data = json.loads(f.read())
|
|
except (OSError, json.JSONDecodeError):
|
|
return make_response({"status": "error"}, 500)
|
|
if not isinstance(data, dict) or field not in data:
|
|
return make_response({"status": "field_not_found"}, 404)
|
|
return make_response({"status": "ok", "data": data[field]}, 200)
|
|
|
|
elif request.method == "POST":
|
|
body = request.get_json(silent=True)
|
|
if not isinstance(body, dict) or "value" not in body:
|
|
return make_response({"status": "bad_request"}, 400)
|
|
value = body["value"]
|
|
if not isinstance(value, str) or len(value) > 4096:
|
|
return make_response({"status": "bad_request"}, 400)
|
|
|
|
existed = os.path.exists(path)
|
|
if not existed:
|
|
Path(path).touch()
|
|
try:
|
|
with open(path, "r") as f:
|
|
data = json.loads(f.read())
|
|
except (OSError, json.JSONDecodeError):
|
|
data = {}
|
|
if not isinstance(data, dict):
|
|
data = {}
|
|
|
|
# PATCH V2 (defense in depth): пароль можно только СОЗДАТЬ при регистрации,
|
|
# но не перезаписать. Функции смены пароля в сервисе нет, поэтому легальный
|
|
# флоу это не ломает, а прямой захват аккаунта через :38002 — закрывает.
|
|
if field == "password" and "password" in data:
|
|
return make_response({"status": "user_exists"}, 409)
|
|
|
|
data[field] = value
|
|
try:
|
|
with open(path, "w") as f:
|
|
f.write(json.dumps(data))
|
|
except OSError:
|
|
return make_response({"status": "error"}, 500)
|
|
return make_response({"status": "ok"}, 200)
|
|
|
|
|
|
@app.route('/images/<string:id>', methods=["GET", "POST"])
|
|
def images(id):
|
|
denied = require_internal()
|
|
if denied is not None:
|
|
return denied
|
|
# PATCH V8: раньше любой нечисловой/несуществующий id давал 500
|
|
try:
|
|
img_id = int(id)
|
|
except (TypeError, ValueError):
|
|
return make_response({"status": "bad_request"}, 400)
|
|
if not (IMG_MIN <= img_id <= IMG_MAX):
|
|
return make_response({"status": "not_found"}, 404)
|
|
path = os.path.join(basic_path, "images", str(img_id).zfill(2))
|
|
|
|
if request.method == 'GET':
|
|
try:
|
|
with open(path, "r") as f:
|
|
data = json.loads(f.read())
|
|
except (OSError, json.JSONDecodeError):
|
|
return make_response({"status": "error"}, 500)
|
|
return make_response({"status": "ok", "data": data}, 200)
|
|
|
|
elif request.method == "POST":
|
|
body = request.get_json(silent=True)
|
|
if not isinstance(body, dict) or "value" not in body:
|
|
return make_response({"status": "bad_request"}, 400)
|
|
value = body["value"]
|
|
if (not isinstance(value, list) or len(value) != 2
|
|
or not all(isinstance(x, str) for x in value)
|
|
or len(value[1]) > 4096):
|
|
return make_response({"status": "bad_request"}, 400)
|
|
try:
|
|
with open(path, "r") as f:
|
|
data = json.loads(f.read())
|
|
if not isinstance(data, dict) or not isinstance(data.get("comments"), list):
|
|
return make_response({"status": "error"}, 500)
|
|
data["comments"].append(value)
|
|
with open(path, "w") as f:
|
|
f.write(json.dumps(data))
|
|
except (OSError, json.JSONDecodeError):
|
|
return make_response({"status": "error"}, 500)
|
|
return make_response({"status": "ok"}, 200)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host="0.0.0.0", port=database_node_port)
|