import os import base64 import mimetypes import logging from datetime import datetime 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() # Configuration from environment variables GITEA_API_URL = os.environ.get("GITEA_API_URL", "https://your-gitea-instance/api/v1") GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "") REPO_OWNER = os.environ.get("REPO_OWNER", "default-owner") REPO_NAME = os.environ.get("REPO_NAME", "default-repo") 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.DEBUG) # Set to INFO for production logger = logging.getLogger(__name__) # Helper: Call Gitea API 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() except requests.exceptions.RequestException as e: logger.error(f"Gitea API request failed: {e}") raise # Helper: Validate IndieAuth token def validate_token(token): try: 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 = { "content": encoded_content, "message": commit_message, "branch": BRANCH, } 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": {}, "actions": ["create", "update", "delete"], }) 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 def create_post(data, user): title = data.get("name", "Untitled Post") content = data.get("content", "") photo = data.get("photo") # Optional: URL of uploaded photo slug = data.get("slug", title.lower().replace(" ", "-")) date = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") md_content = f"""--- title: "{title}" date: {date} author: "{user.get('me')}" --- """ if photo: md_content += f"![Image]({photo})\n\n" md_content += content 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 except Exception as e: logger.error(f"Failed to create post: {e}") return jsonify({"error": "Failed to create post"}), 500 # 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}" ) media_url = f"/{MEDIA_DIR}/{filename}" logger.info(f"Media uploaded: {media_url}") return jsonify({"success": True, "url": media_url}), 201 except Exception as e: 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)