mirror of
https://github.com/NohamR/BrunoServer.git
synced 2026-01-10 00:08:19 +00:00
Add docstrings and improve formatting in server.py
This commit is contained in:
155
server.py
155
server.py
@@ -1,3 +1,4 @@
|
||||
"""Flask server for license activation and verification."""
|
||||
from datetime import datetime
|
||||
import os
|
||||
import uuid
|
||||
@@ -12,105 +13,117 @@ app = Flask(__name__)
|
||||
PENDING_ACTIVATIONS = {}
|
||||
|
||||
def _b64url_encode(data: bytes) -> str:
|
||||
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
||||
"""Base64 URL-safe encode data."""
|
||||
return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=")
|
||||
|
||||
|
||||
def _make_jwt_like(payload: dict) -> str:
|
||||
header = {"alg": "HS256", "typ": "JWT"}
|
||||
header_b64 = _b64url_encode(json.dumps(header, separators=(",", ":"), ensure_ascii=False).encode("utf-8"))
|
||||
payload_b64 = _b64url_encode(json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8"))
|
||||
signature_b64 = _b64url_encode(b"signature")
|
||||
return f"{header_b64}.{payload_b64}.{signature_b64}"
|
||||
"""Create a JWT-like token."""
|
||||
header = {"alg": "HS256", "typ": "JWT"}
|
||||
header_json = json.dumps(header, separators=(",", ":"),
|
||||
ensure_ascii=False).encode("utf-8")
|
||||
header_b64 = _b64url_encode(header_json)
|
||||
payload_json = json.dumps(payload, separators=(",", ":"),
|
||||
ensure_ascii=False).encode("utf-8")
|
||||
payload_b64 = _b64url_encode(payload_json)
|
||||
signature_b64 = _b64url_encode(b"signature")
|
||||
return f"{header_b64}.{payload_b64}.{signature_b64}"
|
||||
|
||||
|
||||
@app.route("/api/v1/license/activate", methods=["POST"])
|
||||
def activate_license():
|
||||
payload = request.get_json(silent=True) or {}
|
||||
print("Received activation request:", payload)
|
||||
"""Handle license activation request."""
|
||||
payload = request.get_json(silent=True) or {}
|
||||
print("Received activation request:", payload)
|
||||
|
||||
license_key = payload.get("licenseKey", "")
|
||||
device_id = payload.get("deviceId", str(uuid.uuid4()))
|
||||
device_name = payload.get("deviceName", "Unnamed Device")
|
||||
email = payload.get("email")
|
||||
license_server_url = payload.get("licenseServerUrl")
|
||||
license_key = payload.get("licenseKey", "")
|
||||
device_id = payload.get("deviceId", str(uuid.uuid4()))
|
||||
device_name = payload.get("deviceName", "Unnamed Device")
|
||||
email = payload.get("email")
|
||||
license_server_url = payload.get("licenseServerUrl")
|
||||
|
||||
activation_id = str(uuid.uuid4())
|
||||
activated_at = datetime.utcnow().isoformat() + "Z"
|
||||
activation_id = str(uuid.uuid4())
|
||||
activated_at = datetime.utcnow().isoformat() + "Z"
|
||||
|
||||
PENDING_ACTIVATIONS[activation_id] = {
|
||||
"licenseKey": license_key,
|
||||
"deviceId": device_id,
|
||||
"deviceName": device_name,
|
||||
"email": email,
|
||||
"licenseServerUrl": license_server_url,
|
||||
"activatedAt": activated_at,
|
||||
}
|
||||
PENDING_ACTIVATIONS[activation_id] = {
|
||||
"licenseKey": license_key,
|
||||
"deviceId": device_id,
|
||||
"deviceName": device_name,
|
||||
"email": email,
|
||||
"licenseServerUrl": license_server_url,
|
||||
"activatedAt": activated_at,
|
||||
}
|
||||
|
||||
resp = {
|
||||
"status": "activated",
|
||||
"licenseKey": license_key,
|
||||
"deviceId": device_id,
|
||||
"deviceName": device_name,
|
||||
"email": email,
|
||||
"activationId": activation_id,
|
||||
"activatedAt": activated_at,
|
||||
}
|
||||
resp = {
|
||||
"status": "activated",
|
||||
"licenseKey": license_key,
|
||||
"deviceId": device_id,
|
||||
"deviceName": device_name,
|
||||
"email": email,
|
||||
"activationId": activation_id,
|
||||
"activatedAt": activated_at,
|
||||
}
|
||||
|
||||
return jsonify(resp), 200
|
||||
return jsonify(resp), 200
|
||||
|
||||
|
||||
@app.route("/api/v1/license/activate/<activation_id>", methods=["POST"])
|
||||
def verify_activation_otp(activation_id: str):
|
||||
payload = request.get_json(silent=True) or {}
|
||||
print("Received OTP verification:", {"activationId": activation_id, **payload})
|
||||
"""Verify activation OTP and return license token."""
|
||||
payload = request.get_json(silent=True) or {}
|
||||
print("Received OTP verification:",
|
||||
{"activationId": activation_id, **payload})
|
||||
|
||||
pending = PENDING_ACTIVATIONS.get(activation_id)
|
||||
if not pending:
|
||||
return jsonify({"error": "Invalid activationId"}), 404
|
||||
pending = PENDING_ACTIVATIONS.get(activation_id)
|
||||
if not pending:
|
||||
return jsonify({"error": "Invalid activationId"}), 404
|
||||
|
||||
now_iso = datetime.utcnow().isoformat() + "Z"
|
||||
license_payload = {
|
||||
"licenseKey": pending.get("licenseKey"),
|
||||
"email": pending.get("email"),
|
||||
"deviceId": pending.get("deviceId"),
|
||||
"deviceName": pending.get("deviceName"),
|
||||
"licenseServerUrl": pending.get("licenseServerUrl"),
|
||||
"plan": "ULTIMATE_EDITION", # ["PRO_EDITION","GOLDEN_EDITION","ULTIMATE_EDITION"]
|
||||
"type": "personal",
|
||||
"createdAt": pending.get("activatedAt"),
|
||||
"updatedAt": now_iso,
|
||||
"trialActive": False,
|
||||
}
|
||||
now_iso = datetime.utcnow().isoformat() + "Z"
|
||||
license_payload = {
|
||||
"licenseKey": pending.get("licenseKey"),
|
||||
"email": pending.get("email"),
|
||||
"deviceId": pending.get("deviceId"),
|
||||
"deviceName": pending.get("deviceName"),
|
||||
"licenseServerUrl": pending.get("licenseServerUrl"),
|
||||
# ["PRO_EDITION","GOLDEN_EDITION","ULTIMATE_EDITION"]
|
||||
"plan": "ULTIMATE_EDITION",
|
||||
"type": "personal",
|
||||
"createdAt": pending.get("activatedAt"),
|
||||
"updatedAt": now_iso,
|
||||
"trialActive": False,
|
||||
}
|
||||
|
||||
token = _make_jwt_like(license_payload)
|
||||
PENDING_ACTIVATIONS.pop(activation_id, None)
|
||||
token = _make_jwt_like(license_payload)
|
||||
PENDING_ACTIVATIONS.pop(activation_id, None)
|
||||
|
||||
return jsonify({
|
||||
"licenseToken": token,
|
||||
}), 200
|
||||
return jsonify({
|
||||
"licenseToken": token,
|
||||
}), 200
|
||||
|
||||
|
||||
@app.route("/api/v1/license/verify", methods=["POST"])
|
||||
def verify_license():
|
||||
payload = request.get_json(silent=True) or {}
|
||||
print("Received license verification:", payload)
|
||||
"""Verify license token."""
|
||||
payload = request.get_json(silent=True) or {}
|
||||
print("Received license verification:", payload)
|
||||
|
||||
response = {
|
||||
"verified": True,
|
||||
"subscription": {
|
||||
"plan": "GOLDEN_EDITION"
|
||||
}
|
||||
}
|
||||
|
||||
return jsonify(response), 200
|
||||
response = {
|
||||
"verified": True,
|
||||
"subscription": {
|
||||
"plan": "GOLDEN_EDITION"
|
||||
}
|
||||
}
|
||||
|
||||
return jsonify(response), 200
|
||||
|
||||
|
||||
def create_app():
|
||||
return app
|
||||
"""Create and return Flask app."""
|
||||
return app
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
host = os.getenv("FLASK_HOST", "127.0.0.1")
|
||||
port = int(os.getenv("FLASK_PORT", "5000"))
|
||||
debug = os.getenv("FLASK_DEBUG", "false").lower() in {"1", "true", "yes"}
|
||||
app.run(host=host, port=port, debug=debug)
|
||||
host = os.getenv("FLASK_HOST", "127.0.0.1")
|
||||
port = int(os.getenv("FLASK_PORT", "5000"))
|
||||
debug = os.getenv("FLASK_DEBUG", "false").lower() in {"1", "true", "yes"}
|
||||
app.run(host=host, port=port, debug=debug)
|
||||
|
||||
Reference in New Issue
Block a user