micropub_server/micropub_server.py

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)