From faaf75d156b4edbc0c5902581e5721d7e77a045d Mon Sep 17 00:00:00 2001 From: Phil Date: Tue, 7 Jan 2025 10:17:48 -0700 Subject: [PATCH] update auth --- micropub_server.py | 65 ++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 60 insertions(+), 5 deletions(-) diff --git a/micropub_server.py b/micropub_server.py index 5a111a5..2bd7150 100644 --- a/micropub_server.py +++ b/micropub_server.py @@ -7,6 +7,7 @@ from flask import Flask, request, jsonify from werkzeug.utils import secure_filename from dotenv import load_dotenv import requests +from urllib.parse import unquote # Load environment variables load_dotenv() @@ -20,12 +21,14 @@ CONTENT_PATH = os.environ.get("CONTENT_PATH", "content") MEDIA_DIR = os.environ.get("MEDIA_DIR", "static/images") BRANCH = os.environ.get("BRANCH", "main") TOKEN_ENDPOINT = os.environ.get("TOKEN_ENDPOINT", "https://tokens.indieauth.com/token") +DOMAIN = os.environ.get("DOMAIN", "https://thesatelliteoflove.com/") +REQUIRED_SCOPES = {"create", "update", "media"} # Initialize Flask app app = Flask(__name__) # Logging configuration -logging.basicConfig(level=logging.INFO) +logging.basicConfig(level=logging.DEBUG) # Set to INFO for production logger = logging.getLogger(__name__) # Helper: Call Gitea API @@ -33,6 +36,7 @@ def gitea_api_request(method, endpoint, data=None): url = f"{GITEA_API_URL}{endpoint}" headers = {"Authorization": f"token {GITEA_TOKEN}"} try: + logger.debug(f"Calling Gitea API: {method} {url}, Data: {data}") response = requests.request(method, url, headers=headers, json=data) response.raise_for_status() return response.json() @@ -43,15 +47,41 @@ def gitea_api_request(method, endpoint, data=None): # Helper: Validate IndieAuth token def validate_token(token): try: - response = requests.post(TOKEN_ENDPOINT, data={"token": token}) - response.raise_for_status() - return response.json() # Should contain "me" (authenticated user's URL) - except requests.exceptions.RequestException as e: + global DOMAIN, REQUIRED_SCOPES # Ensure global variables are accessible + headers = {"Authorization": f"Bearer {token}"} + response = requests.get("https://tokens.indieauth.com/token", headers=headers) + if response.status_code != 200: + logger.error(f"Token validation failed with status {response.status_code}") + return None + + # Parse the x-www-form-urlencoded response + token_data = dict(item.split("=") for item in response.text.split("&")) + logger.debug(f"Raw token data from IndieAuth: {token_data}") + + # Decode URL-encoded values in the response + for key, value in token_data.items(): + token_data[key] = unquote(value) + logger.debug(f"Decoded token data: {token_data}") + + # Validate 'me' field matches DOMAIN + if token_data.get("me") != DOMAIN: + raise ValueError(f"Token 'me' claim ({token_data.get('me')}) does not match the expected domain ({DOMAIN})") + + # Validate required scopes + scopes = token_data.get("scope", "").split("+") # Split by '+' instead of space + if not REQUIRED_SCOPES.issubset(scopes): + raise ValueError(f"Token does not include the required scopes: {REQUIRED_SCOPES}. Found scopes: {scopes}") + + return token_data + except Exception as e: logger.error(f"Token validation failed: {e}") return None + + # Upload content to Gitea def upload_to_gitea(filepath, content, commit_message): + logger.debug(f"Uploading to Gitea: {filepath}, Commit: {commit_message}") encoded_content = base64.b64encode(content.encode() if isinstance(content, str) else content).decode() endpoint = f"/repos/{REPO_OWNER}/{REPO_NAME}/contents/{filepath}" data = { @@ -62,17 +92,22 @@ def upload_to_gitea(filepath, content, commit_message): return gitea_api_request("POST", endpoint, data) # Micropub endpoint +@app.route("/micropub/", methods=["POST", "GET"]) @app.route("/micropub", methods=["POST", "GET"]) def micropub(): + logger.debug(f"Incoming request: {request.method} {request.url}") token = request.headers.get("Authorization", "").replace("Bearer ", "") if not token: + logger.warning("Missing authorization token") return jsonify({"error": "Missing authorization token"}), 401 user = validate_token(token) if not user: + logger.warning("Invalid token") return jsonify({"error": "Invalid token"}), 403 if request.method == "GET": + logger.debug("Micropub discovery request") return jsonify({ "media-endpoint": "/micropub/media", "configurations": {}, @@ -80,9 +115,11 @@ def micropub(): }) data = request.form + logger.debug(f"Micropub POST request data: {data}") if data.get("h") == "entry": return create_post(data, user) + logger.warning("Unsupported Micropub request") return jsonify({"error": "Unsupported Micropub request"}), 400 # Create a new post @@ -105,6 +142,7 @@ author: "{user.get('me')}" filepath = f"{CONTENT_PATH}/{slug}.md" try: + logger.debug(f"Creating post at {filepath} with content:\n{md_content}") response = upload_to_gitea(filepath, md_content, f"Create post: {title}") logger.info(f"Post created: {response['content']['html_url']}") return jsonify({"success": True, "location": response["content"]["html_url"]}), 201 @@ -115,28 +153,35 @@ author: "{user.get('me')}" # Media upload endpoint @app.route("/micropub/media", methods=["POST"]) def media_upload(): + logger.debug(f"Incoming media upload request: {request.url}") token = request.headers.get("Authorization", "").replace("Bearer ", "") if not token: + logger.warning("Missing authorization token") return jsonify({"error": "Missing authorization token"}), 401 user = validate_token(token) if not user: + logger.warning("Invalid token") return jsonify({"error": "Invalid token"}), 403 if "file" not in request.files: + logger.warning("No file provided") return jsonify({"error": "No file provided"}), 400 file = request.files["file"] if file.filename == "": + logger.warning("Empty filename") return jsonify({"error": "Empty filename"}), 400 filename = secure_filename(file.filename) mimetype = mimetypes.guess_type(filename)[0] if mimetype not in ["image/png", "image/jpeg", "image/gif"]: + logger.warning(f"Invalid file type: {mimetype}") return jsonify({"error": "Invalid file type"}), 400 try: + logger.debug(f"Uploading media file: {filename}") response = upload_to_gitea( f"{MEDIA_DIR}/{filename}", file.read(), f"Upload media: {filename}" ) @@ -147,6 +192,16 @@ def media_upload(): logger.error(f"Failed to upload media: {e}") return jsonify({"error": "Failed to upload media"}), 500 +# Health check endpoint +@app.route("/micropub/health", methods=["GET"]) +def health_check(): + """ + Simple health check endpoint to verify that the server is running. + Returns a 200 OK response with a JSON message. + """ + logger.debug("Health check requested") + return jsonify({"status": "ok", "message": "Micropub server is running"}), 200 + # Run the app if __name__ == "__main__": app.run(host="0.0.0.0", port=5000)