Merge pull request 'feat: add 'agent' API key role (content CRUD + backup)' (#4) from feat/agent-role into main

Reviewed-on: #4

api/__init__.py

@@ -1,22 +1,25 @@
-import base64
-import os
-import pkgutil
 from functools import wraps
-
+from datetime import datetime, UTC
 from cryptography import x509
 from cryptography.hazmat.primitives import serialization
 from flask import jsonify, Blueprint, request, make_response
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
 from jwt import decode, ExpiredSignatureError, InvalidTokenError, get_unverified_header
-import importlib
-import requests
 from threading import Lock
+from db.models.APIKey import APIKey
+from db import get_db
 
+import base64
+import os
+import pkgutil
+import secrets
+import string
 import env_provider
 import hashlib
 import json
-
+import importlib
+import requests
 
 _public_key_cache = {}
 _lock = Lock()
@@ -34,7 +37,8 @@ def x5c_to_public_key(x5c):
 
 def get_jwks():
     url = f"{env_provider.KC_HOST}/realms/{env_provider.KC_REALM}/protocol/openid-connect/certs"
-    response = requests.get(url)
+    response = requests.get(url, timeout=5)
+    response.raise_for_status()
     jwks = response.json()
     return jwks
 
@@ -69,7 +73,9 @@ def verify_token(token):
             token,
             public_key,
             algorithms=["RS256"],
-            audience=env_provider.KC_CLIENT_ID
+            audience=env_provider.KC_CLIENT_ID,
+            issuer=f"{env_provider.KC_HOST}/realms/{env_provider.KC_REALM}",
+            options={"require": ["exp", "iat"]},
         )
         return decoded
     except ExpiredSignatureError as e:
@@ -83,7 +89,10 @@ def is_user_admin():
     is_admin = False
     auth_header = request.headers.get('Authorization')
     if auth_header and auth_header.startswith('Bearer'):
-        token = auth_header.split(" ")[1]
+        parts = auth_header.split(" ", 1)
+        if len(parts) != 2 or not parts[1].strip():
+            return is_admin
+        token = parts[1].strip()
         decoded = verify_token(token)
         if decoded:
             user_roles = decoded.get("resource_access", {}).get(env_provider.KC_CLIENT_ID, {}).get("roles", [])
@@ -96,18 +105,45 @@ def require_auth(roles=[]):
         def wrapper(*args, **kwargs):
             if request.method == "OPTIONS":
                 return '', 200
+
             auth_header = request.headers.get('Authorization')
+            api_key_header = request.headers.get('X-API-Key')
+
+            if auth_header and api_key_header:
+                return jsonify({"error": "Cannot use both Bearer token and API Key authentication"}), 403
+            
+            if api_key_header:
+                api_key = get_api_key(api_key_header)
+                if not api_key:
+                    return jsonify({"error": "Invalid API key"}), 401
+
+                expire_time = api_key.expire.replace(tzinfo=UTC) if api_key.expire.tzinfo is None else api_key.expire
+                if datetime.now(UTC) > expire_time:
+                    return jsonify({"error": "API key has expired"}), 401
+
+                if roles and not (set(roles) & set(api_key.roles)):
+                    return jsonify({"error": "Forbidden, permission denied"}), 403
+
+                update_last_used(api_key)
+                return func(*args, **kwargs)
+
             if not auth_header or not auth_header.startswith('Bearer'):
                 return jsonify({"error": "Unauthorized"}), 401
-            token = auth_header.split(" ")[1]
+
+            parts = auth_header.split(" ", 1)
+            if len(parts) != 2 or not parts[1].strip():
+                return jsonify({"error": "Unauthorized"}), 401
+            token = parts[1].strip()
 
             decoded = verify_token(token)
             if not decoded:
                 return jsonify({"error": "Invalid or expired token"}), 401
+
             user_roles = decoded.get("resource_access", {}).get(env_provider.KC_CLIENT_ID, {}).get("roles", [])
             if roles and not (set(roles) & set(user_roles)):
                 print("auth failed")
                 return jsonify({"error": "Forbidden, permission denied"}), 403
+
             print("auth success")
             return func(*args, **kwargs)
         return wrapper
@@ -169,3 +205,43 @@ def etag_response(f):
                 return resp
         return response
     return decorator
+
+
+def generate_api_key(length=32):
+    alphabet = string.ascii_letters + string.digits
+    return ''.join(secrets.choice(alphabet) for _ in range(length))
+
+def get_api_key(key):
+    with get_db() as session:
+        return session.query(APIKey).filter_by(key=key, is_active=True).first()
+
+def update_last_used(api_key):
+    # api_key comes from get_api_key()'s (now closed) session, so it is
+    # detached. Update by key in a fresh session instead of mutating the
+    # detached instance, which would never be persisted.
+    with get_db() as session:
+        session.query(APIKey).filter_by(key=api_key.key).update(
+            {APIKey.last_used_at: datetime.now(UTC)}
+        )
+        session.commit()
+
+
+def get_actor():
+    """Identity string to record as author/last_modified_by.
+
+    - X-API-Key request -> the key's alias
+    - Keycloak Bearer request -> the literal 'admin' (the backend does not
+      track individual KC identities)
+    - otherwise -> None
+    Call only from endpoints already behind @require_auth.
+    """
+    api_key_header = request.headers.get('X-API-Key')
+    if api_key_header:
+        api_key = get_api_key(api_key_header)
+        if api_key:
+            return api_key.alias
+        return None
+    auth_header = request.headers.get('Authorization')
+    if auth_header and auth_header.startswith('Bearer'):
+        return 'admin'
+    return None

api/apikey/__init__.py

@@ -0,0 +1,84 @@
+from datetime import datetime, timedelta, UTC
+from flask import Blueprint, request, jsonify
+from api import generate_api_key
+from db import get_db
+from api import require_auth
+from db.models.APIKey import APIKey
+
+api_key_bp = Blueprint('apikey', __name__, url_prefix='/api/apikey')
+
+# An API key must never be able to request a role broader than what the
+# product defines, regardless of what the request body asks for.
+ALLOWED_API_KEY_ROLES = {'admin', 'creator', 'user', 'agent'}
+
+# Validity window applied on create and on every renewal.
+KEY_TTL = timedelta(days=15)
+
+@api_key_bp.route('/', methods=['POST'])
+@require_auth(roles=['admin'])
+def create_key():
+    """Create an API key, or renew an existing one.
+
+    `alias` is required and unique. Creating with an alias that already
+    exists is treated as a RENEWAL of that key: the same key string is
+    kept (so existing integrations keep working), its validity window is
+    reset, it is reactivated, and name/roles are updated. The (unchanged)
+    key string is returned again.
+    """
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"error": "invalid or missing JSON body"}), 400
+
+    alias = data.get('alias')
+    name = data.get('name')
+    if not alias or not str(alias).strip():
+        return jsonify({"error": "alias is required"}), 400
+    if not name:
+        return jsonify({"error": "Name is required"}), 400
+    alias = str(alias).strip()
+
+    roles = data.get('roles', [])
+    if not isinstance(roles, list) or any(r not in ALLOWED_API_KEY_ROLES for r in roles):
+        return jsonify({"error": f"roles must be a subset of {sorted(ALLOWED_API_KEY_ROLES)}"}), 400
+
+    try:
+        with get_db() as session:
+            existing = session.query(APIKey).filter_by(alias=alias).first()
+            if existing is not None:
+                # Renewal: keep the key string, reset validity, reactivate.
+                existing.name = name
+                existing.roles = roles
+                existing.is_active = True
+                existing.expire = datetime.now(UTC) + KEY_TTL
+                session.commit()
+                result = existing.to_dict()
+                result['renewed'] = True
+                return jsonify(result), 200
+
+            apikey = APIKey(
+                key=generate_api_key(),
+                alias=alias,
+                name=name,
+                roles=roles,
+                expire=datetime.now(UTC) + KEY_TTL,
+            )
+            session.add(apikey)
+            session.commit()
+            result = apikey.to_dict()
+            result['renewed'] = False
+        return jsonify(result), 201
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+@api_key_bp.route('/<key>', methods=['DELETE'])
+@require_auth(roles=['admin'])
+def revoke_key(key):
+    # Query and mutate within the same session, otherwise the update is
+    # performed on a detached instance and silently never persists.
+    with get_db() as session:
+        api_key = session.query(APIKey).filter_by(key=key, is_active=True).first()
+        if not api_key:
+            return jsonify({"error": "API key not found"}), 404
+        api_key.is_active = False
+        session.commit()
+        return jsonify({"message": "API key revoked successfully"}), 200

api/backup.py

@@ -215,6 +215,40 @@ logger = logging.getLogger(__name__)
 
 backup_bp = Blueprint('backup', __name__, url_prefix='/api/backup')
 
+# Upper bounds to defend against zip bombs in uploaded backups.
+_MAX_ARCHIVE_MEMBERS = 50000
+_MAX_UNCOMPRESSED_BYTES = 1 * 1024 * 1024 * 1024  # 1 GiB
+
+
+def safe_extract(zip_ref, dest_dir):
+    """
+    Safely extract a zip archive into dest_dir.
+
+    Prevents "Zip Slip" path traversal (entries like ``../../etc/x`` or
+    absolute paths) and rejects zip bombs by capping the member count and
+    total uncompressed size. Backups are admin-uploaded but may originate
+    from an untrusted source (e.g. the /convert endpoint ingests foreign
+    backups), so the contents must be treated as hostile.
+    """
+    dest_root = os.path.realpath(dest_dir)
+    members = zip_ref.infolist()
+
+    if len(members) > _MAX_ARCHIVE_MEMBERS:
+        raise ValueError("backup archive has too many entries")
+
+    total = 0
+    for member in members:
+        total += member.file_size
+        if total > _MAX_UNCOMPRESSED_BYTES:
+            raise ValueError("backup archive uncompressed size exceeds limit")
+
+        target = os.path.realpath(os.path.join(dest_root, member.filename))
+        if target != dest_root and not target.startswith(dest_root + os.sep):
+            raise ValueError(f"unsafe path in backup archive: {member.filename}")
+
+    zip_ref.extractall(dest_root)
+
+
 def check_and_convert_backup_version(backup_dir):
     """
     Check the backup version and convert it if necessary.
@@ -280,7 +314,7 @@ def convert_backup_endpoint():
         uploaded_file.save(zip_path)
 
         with zipfile.ZipFile(zip_path, 'r') as zip_ref:
-            zip_ref.extractall(backup_dir)
+            safe_extract(zip_ref, backup_dir)
 
         success, error_response = check_and_convert_backup_version(backup_dir)
         if not success:
@@ -312,7 +346,7 @@ def convert_backup_endpoint():
 
 backup_lock = threading.Lock()
 @backup_bp.route('/', methods=['GET'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 def get_backup():
     """
     Create a backup of the application's data.
@@ -331,7 +365,12 @@ def get_backup():
     Response Codes:
         - 200: Backup created successfully
         - 500: Failed to create backup
+        - 429: Another backup operation is in progress
     """
+    if not backup_lock.acquire(blocking=False):
+        return jsonify({"error": "Another backup operation is in progress. Please try again later."}), 429
+
+    original_cwd = os.getcwd()
     try:
         if os.path.exists('Root'):
             shutil.rmtree('Root')
@@ -370,6 +409,11 @@ def get_backup():
     except Exception as e:
         logger.error(f"Failed to get backup: {e}")
         return jsonify({"error": "failed to get backup"}), 500
+    finally:
+        # os.chdir is process-global; always restore CWD so a failure
+        # mid-traverse can't corrupt later requests.
+        os.chdir(original_cwd)
+        backup_lock.release()
 
 
 def create_and_cd(path_name):
@@ -514,7 +558,7 @@ def traverse(path_id, paths):
 
 
 @backup_bp.route('/load', methods=['POST'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 def load_backup():
     """
     Restore data from a backup file.
@@ -552,7 +596,7 @@ def load_backup():
         uploaded_file.save(zip_path)
 
         with zipfile.ZipFile(zip_path, 'r') as zip_ref:
-            zip_ref.extractall(temp_dir)
+            safe_extract(zip_ref, temp_dir)
 
         root_dir = os.path.join(temp_dir, "Root")
         if not os.path.exists(root_dir):

api/log.py

@@ -1,12 +1,18 @@
 #api/log.py
 from flask import Blueprint, jsonify, request
+from api import require_auth
 from db import get_db
 from db.models.Log import Log
 from db.utils import insert_log
 
 logs_bp = Blueprint('log', __name__, url_prefix='/api/log')
 
+# Bound per-entry size so an authenticated-but-low-trust caller can't bloat
+# the log table with multi-megabyte payloads.
+_MAX_LOG_MESSAGE_LEN = 16 * 1024
+
 @logs_bp.route('/', methods=['GET'])
+@require_auth(roles=['admin'])
 def get_logs():
     level = request.args.get('level')
     application = request.args.get('application')
@@ -29,14 +35,17 @@ def get_logs():
     })
 
 @logs_bp.route('/', methods=['POST'])
+@require_auth()
 def create_log():
-    data = request.json
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"error": "invalid or missing JSON body"}), 400
     required_fields = ['level', 'message']
     for field in required_fields:
         if field not in data:
             return jsonify({"error": f"missing {field} in request"}), 400
-    level = data.get('level')
+    level = str(data.get('level'))[:64]
-    message = data.get('message')
+    message = str(data.get('message'))[:_MAX_LOG_MESSAGE_LEN]
     application = "frontend"
     extra = data.get('extra', None)
     log_entry = Log(level=level, message=message, application=application, extra=extra)

api/markdown.py

@@ -1,8 +1,9 @@
 from flask import Blueprint, request, jsonify
 from sqlalchemy import or_
 from api import limiter
-from api import require_auth, etag_response, verify_token, is_user_admin
+from api import require_auth, etag_response, verify_token, is_user_admin, get_actor
 from contexts.RequestContext import RequestContext
+from datetime import datetime, UTC
 from db import get_db
 from db.models.Markdown import Markdown
 from db.models.MarkdownSetting import MarkdownSetting
@@ -15,6 +16,40 @@ logger = logging.getLogger(__name__)
 
 markdown_bp = Blueprint('markdown', __name__, url_prefix='/api/markdown')
 
+
+def _permission_of(session, markdown):
+    """Return the effective permission string ('private'/'protected'/None)."""
+    if markdown.setting_id is None:
+        return None
+    setting = session.query(MarkdownSetting).get(markdown.setting_id)
+    if not setting or not setting.permission_setting_id:
+        return None
+    ps = session.query(MarkdownPermissionSetting).get(setting.permission_setting_id)
+    return ps.permission if ps else None
+
+
+def _filter_visible(session, markdowns, is_admin):
+    """
+    Apply markdown permission settings to a list for non-admin callers:
+    private documents are dropped entirely, protected documents have their
+    content masked. Admins see everything. Mirrors get_markdown()'s checks
+    so listing/search endpoints can no longer leak restricted content.
+    """
+    if is_admin:
+        return [md.to_dict() for md in markdowns]
+
+    visible = []
+    for md in markdowns:
+        permission = _permission_of(session, md)
+        if permission == 'private':
+            continue
+        data = md.to_dict()
+        if permission == 'protected':
+            data['content'] = None
+            data['protected'] = True
+        visible.append(data)
+    return visible
+
 @markdown_bp.route('/', methods=['GET'])
 @limiter.limit(api.get_rate_limit)
 @etag_response
@@ -30,9 +65,10 @@ def get_markdowns():
     Response Codes:
         - 200: Success
     """
+    is_admin = is_user_admin()
     with get_db() as session:
         mds = session.query(Markdown).all()
-    return jsonify([md.to_dict() for md in mds]), 200
+        return jsonify(_filter_visible(session, mds, is_admin)), 200
 
 @markdown_bp.route('/get_home', methods=['GET'])
 @limiter.limit(api.get_rate_limit)
@@ -51,11 +87,15 @@ def get_home():
         - 200: Success
         - 204: No content (home markdown not found)
     """
+    is_admin = is_user_admin()
     with get_db() as session:
         markdown = session.query(Markdown).filter(Markdown.path_id == 1, Markdown.title == "index").first()
         if markdown is None:
             return jsonify({}), 204
-        return jsonify(markdown.to_dict()), 200
+        visible = _filter_visible(session, [markdown], is_admin)
+        if not visible:
+            return jsonify({}), 204
+        return jsonify(visible[0]), 200
 
 @markdown_bp.route('/by_path/<int:path_id>', methods=['GET'])
 @limiter.limit(api.get_rate_limit)
@@ -75,9 +115,10 @@ def get_markdowns_by_path(path_id):
     Response Codes:
         - 200: Success
     """
+    is_admin = is_user_admin()
     with get_db() as session:
         markdowns = session.query(Markdown).filter(Markdown.path_id == path_id).all()
-        return jsonify([md.to_dict() for md in markdowns]), 200
+        return jsonify(_filter_visible(session, markdowns, is_admin)), 200
 
 @markdown_bp.route('/get_index/<int:path_id>', methods=['GET'])
 @limiter.limit(api.get_rate_limit)
@@ -99,11 +140,15 @@ def get_index(path_id):
         - 200: Success
         - 204: No content (index markdown not found)
     """
+    is_admin = is_user_admin()
     with get_db() as session:
         markdown = session.query(Markdown).filter(Markdown.path_id == path_id).filter(Markdown.title == "index").first()
         if markdown is None:
             return jsonify({}), 204
-        return jsonify(markdown.to_dict()), 200
+        visible = _filter_visible(session, [markdown], is_admin)
+        if not visible:
+            return jsonify({}), 204
+        return jsonify(visible[0]), 200
 
 
 
@@ -149,7 +194,7 @@ def get_markdown(markdown_id):
         return jsonify(markdown.to_dict()), 200
 
 @markdown_bp.route('/', methods=['POST'])
-@require_auth(roles=['admin', 'creator'])
+@require_auth(roles=['admin', 'creator', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def create_markdown():
     """
@@ -181,7 +226,13 @@ def create_markdown():
     setting_id = data.get('setting_id', None)
     if not title or not content:
         return jsonify({"error": "missing required fields"}), 400
-    new_markdown = Markdown(title=title, content=content, path_id=path_id, shortcut=shortcut, setting_id=setting_id)
+    actor = get_actor()
+    now = datetime.now(UTC)
+    new_markdown = Markdown(
+        title=title, content=content, path_id=path_id, shortcut=shortcut,
+        setting_id=setting_id, author=actor, last_modified_by=actor,
+        created_at=now, updated_at=now,
+    )
     with get_db() as session:
         try:
             if shortcut != "":
@@ -199,7 +250,7 @@ def create_markdown():
             return jsonify({"error": f"create failed - {errno}"}), 500
 
 @markdown_bp.route('/<int:markdown_id>', methods=['PUT', 'PATCH'])
-@require_auth(roles=['admin', 'creator'])
+@require_auth(roles=['admin', 'creator', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def update_markdown(markdown_id):
     """
@@ -257,18 +308,21 @@ def update_markdown(markdown_id):
                 markdown.shortcut = data.get('shortcut')
             if 'setting_id' in data:
                 markdown.setting_id = data.get('setting_id')
+        markdown.updated_at = datetime.now(UTC)
+        markdown.last_modified_by = get_actor()
         session.commit()
         markdown_updated.send(None, payload=markdown.to_dict())
         return jsonify(markdown.to_dict()), 200
 
 @markdown_bp.route('/<int:markdown_id>', methods=['DELETE'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def delete_markdown(markdown_id):
     """
     Delete a markdown document.
 
-    This endpoint deletes an existing markdown document.
+    This endpoint deletes an existing markdown document and cascades the deletion
+    to related settings to avoid foreign key conflicts.
     It requires authentication with the 'admin' role.
 
     Request:
@@ -280,22 +334,64 @@ def delete_markdown(markdown_id):
     Response Codes:
         - 200: Deleted successfully
         - 404: Markdown not found
+        - 500: Server error during cascade deletion
     """
     with get_db() as session:
+        try:
             markdown = session.get(Markdown, markdown_id)
             if markdown is None:
                 logger.error(f"failed to delete markdown: {markdown_id}")
                 errno = RequestContext.get_error_id()
                 return jsonify({"error": f"file not found - {errno}"}), 404
+            
             md = markdown.to_dict()
+            
+
+            if markdown.setting_id:
+                markdown_setting = session.query(MarkdownSetting).get(markdown.setting_id)
+                if markdown_setting:
+                    template_setting_id = markdown_setting.template_setting_id
+                    permission_setting_id = markdown_setting.permission_setting_id
+
+                    markdown_setting.template_setting_id = None
+                    markdown_setting.permission_setting_id = None
+                    session.flush()
+
+                    if template_setting_id:
+                        from db.models.MarkdownTemplateSetting import MarkdownTemplateSetting
+                        template_setting = session.query(MarkdownTemplateSetting).get(template_setting_id)
+                        if template_setting:
+                            session.delete(template_setting)
+                    
+                    if permission_setting_id:
+                        permission_setting = session.query(MarkdownPermissionSetting).get(permission_setting_id)
+                        if permission_setting:
+                            session.delete(permission_setting)
+
+                    session.delete(markdown_setting)
+
+            # Send webhook event before committing the transaction
+            # This ensures webhook handlers can still access related data
+            markdown_deleted.send(None, payload=md)
+            
             session.delete(markdown)
             session.commit()
-        markdown_deleted.send(None, payload=md)
+            
+            logger.info(f"Successfully deleted markdown {markdown_id} with cascade deletion")
             return jsonify(md), 200
             
+        except Exception as e:
+            import traceback
+            logger.error(f"Failed to delete markdown {markdown_id}: {e}")
+            logger.error(f"Exception type: {type(e).__name__}")
+            logger.error(f"Full traceback:\n{traceback.format_exc()}")
+            errno = RequestContext.get_error_id()
+            session.rollback()
+            return jsonify({"error": f"delete failed - {errno}"}), 500
+
 
 @markdown_bp.route('/move_forward/<int:markdown_id>', methods=['PATCH'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def move_forward(markdown_id):
     """
@@ -332,7 +428,7 @@ def move_forward(markdown_id):
 
 
 @markdown_bp.route('/move_backward/<int:markdown_id>', methods=['PATCH'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def move_backward(markdown_id):
     """
@@ -386,11 +482,19 @@ def search_markdowns(keyword):
     Response Codes:
         - 200: Success
     """
+    is_admin = is_user_admin()
+    # Escape LIKE metacharacters so user input can't inject wildcards
+    # (full-table-scan DoS) and so search actually matches substrings.
+    escaped = keyword.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')
+    pattern = f"%{escaped}%"
     with get_db() as session:
         res = session.query(Markdown).filter(
-            or_(Markdown.title.like(keyword), Markdown.content.like(keyword))
+            or_(
-        ).all()
+                Markdown.title.like(pattern, escape='\\'),
-        return jsonify([md.to_dict() for md in res]), 200
+                Markdown.content.like(pattern, escape='\\'),
+            )
+        ).limit(200).all()
+        return jsonify(_filter_visible(session, res, is_admin)), 200
 
 @markdown_bp.route('/links', methods=['GET'])
 @limiter.limit(api.get_rate_limit)

api/patch.py

@@ -0,0 +1,131 @@
+from datetime import datetime, UTC
+from flask import Blueprint, request, jsonify
+from api import limiter, require_auth, is_user_admin, get_actor
+from contexts.RequestContext import RequestContext
+from db import get_db
+from db.models.Markdown import Markdown
+from db.models.MarkdownPatch import MarkdownPatch
+from db.models.MarkdownSetting import MarkdownSetting
+from db.models.MarkdownPermissionSetting import MarkdownPermissionSetting
+import api
+import logging
+
+logger = logging.getLogger(__name__)
+
+patch_bp = Blueprint('patch', __name__, url_prefix='/api/patch')
+
+
+def _parent_visible_to_caller(session, markdown, is_admin):
+    """Patch cards inherit the parent markdown's visibility. Admins see all;
+    non-admins are denied for private or protected parents (so patches can't
+    be used to leak restricted content)."""
+    if is_admin or markdown.setting_id is None:
+        return True
+    setting = session.query(MarkdownSetting).get(markdown.setting_id)
+    if not setting or not setting.permission_setting_id:
+        return True
+    ps = session.query(MarkdownPermissionSetting).get(setting.permission_setting_id)
+    if ps and ps.permission in ('private', 'protected'):
+        return False
+    return True
+
+
+@patch_bp.route('/by_markdown/<int:markdown_id>', methods=['GET'])
+@limiter.limit(api.get_rate_limit)
+def get_patches(markdown_id):
+    """List patch cards for a markdown, ordered, respecting parent visibility."""
+    is_admin = is_user_admin()
+    with get_db() as session:
+        markdown = session.query(Markdown).get(markdown_id)
+        if markdown is None:
+            return jsonify({"error": "markdown not found"}), 404
+        if not _parent_visible_to_caller(session, markdown, is_admin):
+            return jsonify({"error": "permission denied"}), 403
+        patches = (
+            session.query(MarkdownPatch)
+            .filter(MarkdownPatch.markdown_id == markdown_id)
+            .order_by(MarkdownPatch.order.asc(), MarkdownPatch.id.asc())
+            .all()
+        )
+        return jsonify([p.to_dict() for p in patches]), 200
+
+
+@patch_bp.route('/', methods=['POST'])
+@require_auth(roles=['admin', 'creator', 'agent'])
+@limiter.limit(api.get_rate_limit)
+def create_patch():
+    """Create a patch card. Body: markdown_id, content, title?, order?"""
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"error": "invalid or missing JSON body"}), 400
+    markdown_id = data.get('markdown_id')
+    content = data.get('content')
+    if not markdown_id or content is None or content == "":
+        return jsonify({"error": "missing required fields"}), 400
+    with get_db() as session:
+        if session.query(Markdown).get(markdown_id) is None:
+            return jsonify({"error": "markdown not found"}), 404
+        try:
+            actor = get_actor()
+            now = datetime.now(UTC)
+            patch = MarkdownPatch(
+                markdown_id=markdown_id,
+                title=data.get('title'),
+                content=content,
+                order=data.get('order', 0),
+                author=actor,
+                last_modified_by=actor,
+                created_at=now,
+                updated_at=now,
+            )
+            session.add(patch)
+            session.commit()
+            return jsonify(patch.to_dict()), 201
+        except Exception as e:
+            logger.error(f"failed to create patch: {e}")
+            errno = RequestContext.get_error_id()
+            session.rollback()
+            return jsonify({"error": f"create failed - {errno}"}), 500
+
+
+@patch_bp.route('/<int:patch_id>', methods=['PUT', 'PATCH'])
+@require_auth(roles=['admin', 'creator', 'agent'])
+@limiter.limit(api.get_rate_limit)
+def update_patch(patch_id):
+    """Update a patch card (title/content/order)."""
+    data = request.get_json(silent=True)
+    if not data:
+        return jsonify({"error": "invalid or missing JSON body"}), 400
+    with get_db() as session:
+        patch = session.query(MarkdownPatch).get(patch_id)
+        if patch is None:
+            return jsonify({"error": "patch not found"}), 404
+        if request.method == "PUT":
+            patch.title = data.get('title')
+            patch.content = data.get('content')
+            patch.order = data.get('order', 0)
+        else:
+            if 'title' in data:
+                patch.title = data.get('title')
+            if 'content' in data:
+                patch.content = data.get('content')
+            if 'order' in data:
+                patch.order = data.get('order')
+        patch.updated_at = datetime.now(UTC)
+        patch.last_modified_by = get_actor()
+        session.commit()
+        return jsonify(patch.to_dict()), 200
+
+
+@patch_bp.route('/<int:patch_id>', methods=['DELETE'])
+@require_auth(roles=['admin', 'agent'])
+@limiter.limit(api.get_rate_limit)
+def delete_patch(patch_id):
+    """Delete a patch card."""
+    with get_db() as session:
+        patch = session.query(MarkdownPatch).get(patch_id)
+        if patch is None:
+            return jsonify({"error": "patch not found"}), 404
+        session.delete(patch)
+        session.commit()
+        return jsonify({"message": "patch deleted"}), 200

api/path.py

@@ -82,7 +82,7 @@ def get_path_by_parent(parent_id):
 
 @path_bp.route('/', methods=['POST'])
 @limiter.limit(api.get_rate_limit)
-@require_auth(roles=['admin', 'creator'])
+@require_auth(roles=['admin', 'creator', 'agent'])
 def create_path():
     """
     Create a new path.
@@ -119,7 +119,7 @@ def create_path():
 
 @path_bp.route('/<int:path_id>', methods=['PUT'])
 @limiter.limit(api.get_rate_limit)
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 def update_path(path_id):
     """
     Update a path.
@@ -158,7 +158,7 @@ def update_path(path_id):
 
 @path_bp.route('/<int:path_id>', methods=['PATCH'])
 @limiter.limit(api.get_rate_limit)
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 def patch_path(path_id):
     """
     Partially update a path.
@@ -205,7 +205,7 @@ def patch_path(path_id):
 
 @path_bp.route('/<int:path_id>', methods=['DELETE'])
 @limiter.limit(api.get_rate_limit)
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 def delete_path(path_id):
     """
     Delete a path.
@@ -240,7 +240,7 @@ def delete_path(path_id):
 
 
 @path_bp.route('/move_forward/<int:path_id>', methods=['PATCH'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def move_forward(path_id):
     """
@@ -277,7 +277,7 @@ def move_forward(path_id):
 
 
 @path_bp.route('/move_backward/<int:path_id>', methods=['PATCH'])
-@require_auth(roles=['admin'])
+@require_auth(roles=['admin', 'agent'])
 @limiter.limit(api.get_rate_limit)
 def move_backward(path_id):
     """

api/setting/markdown/__init__.py

@@ -1,13 +1,12 @@
-from flask import jsonify, request
+from flask import jsonify, request, Blueprint
 
 import api
 from api import limiter, etag_response, require_auth
-from api.setting import setting_bp
 from db import get_db
 from db.models.MarkdownSetting import MarkdownSetting
 
-
+markdown_setting_bp = Blueprint('markdown_setting', __name__, url_prefix='/api/setting/markdown')
-@setting_bp.route('/markdown/<int:setting_id>', methods=['GET'])
+@markdown_setting_bp.route('/<int:setting_id>', methods=['GET'])
 @limiter.limit(api.get_rate_limit)
 @etag_response
 def get_markdown_path(setting_id):
@@ -33,7 +32,7 @@ def get_markdown_path(setting_id):
         return jsonify(setting.to_dict()), 200
 
 
-@setting_bp.route('/markdown/', methods=['POST'])
+@markdown_setting_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_markdown_setting():
     """
@@ -68,7 +67,7 @@ def create_markdown_setting():
     except Exception:
         return jsonify({"error": "failed to create setting"}), 500
 
-@setting_bp.route('/markdown/<int:setting_id>', methods=['PUT', 'PATCH'])
+@markdown_setting_bp.route('/<int:setting_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_markdown_setting(setting_id):
     """
@@ -106,7 +105,7 @@ def update_markdown_setting(setting_id):
         return jsonify({"error": "failed to update setting"}), 500
 
 
-@setting_bp.route('/markdown/<int:setting_id>', methods=['DELETE'])
+@markdown_setting_bp.route('/<int:setting_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_markdown_setting(setting_id):
     """

api/setting/markdown/permission.py

@@ -1,12 +1,11 @@
-from flask import jsonify, request
+from flask import jsonify, request, Blueprint
 import api
 from api import etag_response, limiter, require_auth
-from api.setting import setting_bp
 from db import get_db
 from db.models.MarkdownPermissionSetting import MarkdownPermissionSetting
 
-
+markdown_permission_setting_bp = Blueprint('markdown_permission_setting', __name__, url_prefix='/api/setting/markdown/permission')
-@setting_bp.route('/markdown/permission/<int:setting_id>/', methods=['GET'])
+@markdown_permission_setting_bp.route('/<int:setting_id>', methods=['GET'])
 @etag_response
 @limiter.limit(api.get_rate_limit)
 def get_permission_setting(setting_id):
@@ -32,7 +31,7 @@ def get_permission_setting(setting_id):
         return jsonify(setting.to_dict()), 200
 
 
-@setting_bp.route('/markdown/permission/', methods=['POST'])
+@markdown_permission_setting_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_permission_setting():
     """
@@ -59,7 +58,7 @@ def create_permission_setting():
         return jsonify(new_setting.to_dict()), 201
 
 
-@setting_bp.route('/markdown/permission/<int:setting_id>', methods=['PUT', 'PATCH'])
+@markdown_permission_setting_bp.route('/<int:setting_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_permission_setting(setting_id):
     """
@@ -95,7 +94,7 @@ def update_permission_setting(setting_id):
         session.commit()
         return jsonify(setting.to_dict()), 200
 
-@setting_bp.route('/markdown/permission/<int:setting_id>', methods=['DELETE'])
+@markdown_permission_setting_bp.route('/<int:setting_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_permission_setting(setting_id):
     """

api/setting/markdown/template.py

@@ -1,12 +1,12 @@
-from flask import jsonify, request
+from flask import jsonify, request, Blueprint
 import api
 from api import etag_response, limiter, require_auth
-from api.setting import setting_bp
 from db import get_db
 from db.models.MarkdownTemplateSetting import MarkdownTemplateSetting
 
 
-@setting_bp.route('/markdown/template/', methods=['GET'])
+markdown_template_setting_bp = Blueprint('markdown_template_setting', __name__, url_prefix='/api/setting/markdown/template')
+@markdown_template_setting_bp.route('/', methods=['GET'])
 @etag_response
 def list_template_settings():
     """
@@ -24,7 +24,7 @@ def list_template_settings():
         settings = session.query(MarkdownTemplateSetting).all()
         return jsonify([s.to_dict() for s in settings]), 200
 
-@setting_bp.route('/markdown/template/<int:setting_id>/', methods=['GET'])
+@markdown_template_setting_bp.route('/<int:setting_id>', methods=['GET'])
 @etag_response
 @limiter.limit(api.get_rate_limit)
 def get_template_setting(setting_id):
@@ -50,7 +50,7 @@ def get_template_setting(setting_id):
         return jsonify(setting.to_dict()), 200
 
 
-@setting_bp.route('/markdown/template/', methods=['POST'])
+@markdown_template_setting_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_template_setting():
     """
@@ -77,7 +77,7 @@ def create_template_setting():
         return jsonify(new_setting.to_dict()), 201
 
 
-@setting_bp.route('/markdown/template/<int:setting_id>', methods=['PUT', 'PATCH'])
+@markdown_template_setting_bp.route('/<int:setting_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_template_setting(setting_id):
     """
@@ -113,7 +113,7 @@ def update_template_setting(setting_id):
         session.commit()
         return jsonify(setting.to_dict()), 200
 
-@setting_bp.route('/markdown/template/<int:setting_id>', methods=['DELETE'])
+@markdown_template_setting_bp.route('/<int:setting_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_template_setting(setting_id):
     """

api/setting/path/__init__.py

@@ -1,15 +1,16 @@
-from flask import jsonify, request
+from flask import jsonify, request, Blueprint
 
 import api
 from api import limiter, require_auth, etag_response
-from api.setting import setting_bp
 from db import get_db
 from db.models.PathSetting import PathSetting
 
 import logging
 logger = logging.getLogger(__name__)
 
-@setting_bp.route('/path/<int:setting_id>', methods=['GET'])
+path_setting_bp = Blueprint('path_setting', __name__, url_prefix='/api/setting/path')
+
+@path_setting_bp.route('/<int:setting_id>', methods=['GET'])
 @limiter.limit(api.get_rate_limit)
 @etag_response
 def get_path_setting(setting_id):
@@ -34,7 +35,7 @@ def get_path_setting(setting_id):
             return jsonify({}), 204
         return jsonify(setting.to_dict()), 200
 
-@setting_bp.route('/path/', methods=['POST'])
+@path_setting_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_path_setting():
     """
@@ -71,7 +72,7 @@ def create_path_setting():
             return jsonify({"error": "failed to create setting"}), 500
 
 
-@setting_bp.route('/path/<int:setting_id>', methods=['PUT', 'PATCH'])
+@path_setting_bp.route('/<int:setting_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_path_setting(setting_id):
     """
@@ -108,7 +109,7 @@ def update_path_setting(setting_id):
         except Exception:
             return jsonify({"error": "failed to update path setting"}), 500
 
-@setting_bp.route('/path/<int:setting_id>', methods=['DELETE'])
+@path_setting_bp.route('/<int:setting_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_path_setting(setting_id):
     """

api/setting/path/webhook.py

@@ -1,11 +1,10 @@
-from flask import jsonify, request
+from flask import jsonify, request, Blueprint
 from api import require_auth
-from api.setting import setting_bp
 from db import get_db
 from db.models.WebhookSetting import WebhookSetting
 
-
+path_webhook_setting_bp = Blueprint('path_webhook_setting', __name__, url_prefix='/api/setting/path/webhook')
-@setting_bp.route('/path/webhook/', methods=['GET'])
+@path_webhook_setting_bp.route('/', methods=['GET'])
 @require_auth(roles=['admin'])
 def list_webhook_settings():
     """
@@ -25,7 +24,7 @@ def list_webhook_settings():
         return jsonify([s.to_dict() for s in settings]), 200
 
 
-@setting_bp.route('/path/webhook/<int:setting_id>', methods=['GET'])
+@path_webhook_setting_bp.route('/<int:setting_id>', methods=['GET'])
 @require_auth(roles=['admin'])
 def webhook_setting(setting_id):
     """
@@ -51,7 +50,7 @@ def webhook_setting(setting_id):
         return jsonify(setting.to_dict()), 200
 
 
-@setting_bp.route('/path/webhook/', methods=['POST'])
+@path_webhook_setting_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_webhook_setting():
     """
@@ -87,7 +86,7 @@ def create_webhook_setting():
         return jsonify(setting.to_dict()), 201
 
 
-@setting_bp.route('/path/webhook/<int:setting_id>', methods=['PUT', 'PATCH'])
+@path_webhook_setting_bp.route('/<int:setting_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_webhook_setting(setting_id):
     """
@@ -131,7 +130,7 @@ def update_webhook_setting(setting_id):
         return jsonify(setting.to_dict()), 200
 
 
-@setting_bp.route('/path/webhook/<int:setting_id>', methods=['DELETE'])
+@path_webhook_setting_bp.route('/<int:setting_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_webhook_setting(setting_id):
     """

api/template/markdown/__init__.py

@@ -1,10 +1,12 @@
-from flask import jsonify, request
+from flask import jsonify, request, Blueprint
 from api import etag_response, require_auth
-from api.template import template_bp
 from db import get_db
 from db.models.MarkdownTemplate import MarkdownTemplate
 
 cached_templates = {}
+
+markdown_template_bp = Blueprint('markdown_template', __name__, url_prefix='/api/template/markdown')
+
 def inflate_template(template):
     for parameter in template.get('parameters'):
         if parameter.get('type', {}).get('base_type') == 'template':
@@ -19,7 +21,7 @@ def inflate_template(template):
     return template
 
 
-@template_bp.route('/markdown/<int:template_id>', methods=['GET'])
+@markdown_template_bp.route('/<int:template_id>', methods=['GET'])
 @etag_response
 def get_markdown_template(template_id):
     """
@@ -42,10 +44,9 @@ def get_markdown_template(template_id):
         template = session.query(MarkdownTemplate).get(template_id)
         if template is None:
             return jsonify({}), 204
-        print(inflate_template(template.to_dict()))
         return jsonify(inflate_template(template.to_dict())), 200
 
-@template_bp.route('/markdown/', methods=['GET'])
+@markdown_template_bp.route('/', methods=['GET'])
 @etag_response
 def get_markdown_templates():
     """
@@ -62,11 +63,10 @@ def get_markdown_templates():
     """
     with get_db() as session:
         templates = session.query(MarkdownTemplate).all()
-        print(templates)
         return jsonify([inflate_template(template.to_dict()) for template in templates]), 200
 
 
-@template_bp.route('/markdown/', methods=['POST'])
+@markdown_template_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_markdown_template():
     """
@@ -103,7 +103,7 @@ def create_markdown_template():
         return jsonify({"error": "failed to create markdown template"}), 400
 
 
-@template_bp.route('/markdown/<int:template_id>', methods=['PUT', 'PATCH'])
+@markdown_template_bp.route('/<int:template_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_markdown_template(template_id):
     """
@@ -142,7 +142,7 @@ def update_markdown_template(template_id):
         return jsonify(template.to_dict()), 200
 
 
-@template_bp.route('/markdown/<int:template_id>', methods=['DELETE'])
+@markdown_template_bp.route('/<int:template_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_markdown_template(template_id):
     """

api/template/path/__init__.py

@@ -1,12 +1,13 @@
 from flask import jsonify, request
+from flask.sansio.blueprints import Blueprint
 
 from api import etag_response, require_auth
 from api.template import template_bp
 from db import get_db
 from db.models.PathTemplate import PathTemplate
 
-
+path_template_bp = Blueprint('path_template', __name__, url_prefix='/api/template/path')
-@template_bp.route('/path/<int:template_id>', methods=['GET'])
+@path_template_bp.route('/<int:template_id>', methods=['GET'])
 @etag_response
 def get_path_template(template_id):
     """
@@ -30,7 +31,7 @@ def get_path_template(template_id):
             return jsonify({}), 204
         return jsonify(template.to_dict()), 200
 
-@template_bp.route('/path/', methods=['GET'])
+@path_template_bp.route('/', methods=['GET'])
 @etag_response
 def get_path_templates():
     """
@@ -49,7 +50,7 @@ def get_path_templates():
         return jsonify([template.to_dict() for template in templates]), 200
 
 
-@template_bp.route('/path/', methods=['POST'])
+@path_template_bp.route('/', methods=['POST'])
 @require_auth(roles=['admin'])
 def create_path_template():
     """
@@ -84,7 +85,7 @@ def create_path_template():
         return jsonify({"error": "failed to create path template"}), 400
 
 
-@template_bp.route('/path/<int:template_id>', methods=['PUT', 'PATCH'])
+@path_template_bp.route('/<int:template_id>', methods=['PUT', 'PATCH'])
 @require_auth(roles=['admin'])
 def update_path_template(template_id):
     """
@@ -118,7 +119,7 @@ def update_path_template(template_id):
         return jsonify(template.to_dict()), 200
 
 
-@template_bp.route('/path/<int:template_id>', methods=['DELETE'])
+@path_template_bp.route('/<int:template_id>', methods=['DELETE'])
 @require_auth(roles=['admin'])
 def delete_path_template(template_id):
     """

apikey_cli.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""Admin CLI for API key management — no HTTP, no admin login.
+
+Operates directly on the database (same env as the backend), so it must
+run where the DB is reachable, e.g. inside the backend container:
+
+    docker compose exec backend python apikey_cli.py create \
+        --alias ci-bot --name "CI bot" --roles creator
+
+    docker compose exec backend python apikey_cli.py list
+    docker compose exec backend python apikey_cli.py revoke --alias ci-bot
+
+`create` with an existing --alias renews that key (same key string,
+validity reset, reactivated, name/roles updated) — matching the HTTP
+POST /api/apikey behaviour.
+"""
+import argparse
+import secrets
+import string
+import sys
+from datetime import datetime, timedelta, UTC
+
+from db import get_db
+from db.models.APIKey import APIKey
+
+# Keep in sync with api.apikey.ALLOWED_API_KEY_ROLES
+ALLOWED_ROLES = {"admin", "creator", "user", "agent"}
+KEY_TTL_DEFAULT_DAYS = 15
+
+
+def _gen_key(length=32):
+    alphabet = string.ascii_letters + string.digits
+    return "".join(secrets.choice(alphabet) for _ in range(length))
+
+
+def _validate_roles(roles):
+    bad = [r for r in roles if r not in ALLOWED_ROLES]
+    if bad:
+        sys.exit(f"error: invalid role(s) {bad}; allowed: {sorted(ALLOWED_ROLES)}")
+
+
+def cmd_create(args):
+    alias = args.alias.strip()
+    if not alias:
+        sys.exit("error: --alias is required")
+    roles = args.roles or []
+    _validate_roles(roles)
+    expire = datetime.now(UTC) + timedelta(days=args.ttl_days)
+
+    with get_db() as session:
+        existing = session.query(APIKey).filter_by(alias=alias).first()
+        if existing is not None:
+            existing.name = args.name
+            existing.roles = roles
+            existing.is_active = True
+            existing.expire = expire
+            session.commit()
+            row, renewed = existing.to_dict(), True
+        else:
+            ak = APIKey(
+                key=_gen_key(), alias=alias, name=args.name,
+                roles=roles, expire=expire,
+            )
+            session.add(ak)
+            session.commit()
+            row, renewed = ak.to_dict(), False
+
+    print("renewed" if renewed else "created")
+    print(f"  alias : {row['alias']}")
+    print(f"  name  : {row['name']}")
+    print(f"  roles : {row['roles']}")
+    print(f"  expire: {row['expire']}")
+    print(f"  key   : {row['key']}")
+
+
+def cmd_list(args):
+    with get_db() as session:
+        keys = session.query(APIKey).order_by(APIKey.created_at).all()
+        rows = [k.to_dict() for k in keys]
+    if not rows:
+        print("(no API keys)")
+        return
+    for r in rows:
+        key = r["key"] if args.show_keys else (r["key"][:6] + "…")
+        state = "active" if r["is_active"] else "revoked"
+        print(
+            f"{r['alias']!r:<22} {state:<8} roles={r['roles']} "
+            f"expire={r['expire']} last_used={r['last_used_at']} "
+            f"name={r['name']!r} key={key}"
+        )
+
+
+def cmd_revoke(args):
+    with get_db() as session:
+        q = session.query(APIKey)
+        ak = (q.filter_by(alias=args.alias).first() if args.alias
+              else q.filter_by(key=args.key).first())
+        if ak is None:
+            sys.exit("error: API key not found")
+        ak.is_active = False
+        session.commit()
+        print(f"revoked: alias={ak.alias} name={ak.name!r}")
+
+
+def main():
+    p = argparse.ArgumentParser(prog="apikey_cli", description=__doc__)
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    c = sub.add_parser("create", help="create or renew (by alias) an API key")
+    c.add_argument("--alias", required=True, help="unique alias; reuse to renew")
+    c.add_argument("--name", required=True, help="human-readable name")
+    c.add_argument("--roles", nargs="*", default=[],
+                   help=f"subset of {sorted(ALLOWED_ROLES)}")
+    c.add_argument("--ttl-days", type=int, default=KEY_TTL_DEFAULT_DAYS,
+                   dest="ttl_days", help="validity window in days")
+    c.set_defaults(func=cmd_create)
+
+    l = sub.add_parser("list", help="list all API keys")
+    l.add_argument("--show-keys", action="store_true",
+                   help="print full key strings (default: masked)")
+    l.set_defaults(func=cmd_list)
+
+    r = sub.add_parser("revoke", help="deactivate an API key")
+    g = r.add_mutually_exclusive_group(required=True)
+    g.add_argument("--alias", help="revoke by alias")
+    g.add_argument("--key", help="revoke by key string")
+    r.set_defaults(func=cmd_revoke)
+
+    args = p.parse_args()
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()

app.py

@@ -1,4 +1,6 @@
 from pprint import pprint
+
+from events.WebhookEventHandlers import register_all_webhook_event_handlers
 from logging_handlers.DatabaseLogHandler import DatabaseLogHandler
 from api import limiter
 from flask import Flask, request
@@ -33,7 +35,8 @@ CORS(app, resources={
             r"https?://localhost:\d+",
             r"https?://127\.0\.0\.1:\d+",
             r"https?://localhost"
-        ]
+        ],
+        "supports_credentials": True
     }
 },
      expose_headers=['Content-Disposition']
@@ -42,7 +45,7 @@ CORS(app, resources={
 limiter.init_app(app)
 
 api.register_blueprints(app)
-
+register_all_webhook_event_handlers()
 @app.before_request
 def log_request():
     if request.path.startswith("/api/log"):
@@ -50,7 +53,12 @@ def log_request():
     logger.info(f"Request received: {request.method} {request.path} from {request.remote_addr}")
 
 
+api.init_rate_limits(app)
+
 if __name__ == '__main__':
-    api.init_rate_limits(app)
+    print("env")
     pprint(env_provider.summerize())
-    app.run(host='0.0.0.0', port=5000, debug=True, use_reloader=True)
+    # The Werkzeug debugger allows remote code execution. Only enable it in
+    # an explicit dev environment, never in the published production image.
+    is_dev = env_provider.ENVIRONMENT == "dev"
+    app.run(host='0.0.0.0', port=5000, debug=is_dev, use_reloader=is_dev)

db/__init__.py

@@ -59,12 +59,71 @@ def init_payload():
         session.commit()
 
 
+def _column_exists(conn, table, column):
+    row = conn.execute(text(
+        "SELECT 1 FROM information_schema.columns "
+        "WHERE table_schema = :db AND table_name = :t AND column_name = :c"
+    ), {"db": DB_NAME, "t": table, "c": column}).first()
+    return row is not None
+
+
+def _index_exists(conn, table, index):
+    row = conn.execute(text(
+        "SELECT 1 FROM information_schema.statistics "
+        "WHERE table_schema = :db AND table_name = :t AND index_name = :i"
+    ), {"db": DB_NAME, "t": table, "i": index}).first()
+    return row is not None
+
+
+def run_migrations():
+    """Idempotent additive schema migrations for already-existing tables.
+
+    create_all() creates missing tables (with the new columns) for a fresh
+    DB, but never alters existing ones. This adds the new columns to legacy
+    tables and backfills sensible defaults. Safe to run on every startup.
+    """
+    # (table, column, DDL, backfill SQL or None)
+    steps = [
+        ("apikey", "alias", "ALTER TABLE apikey ADD COLUMN alias VARCHAR(255) NULL",
+         "UPDATE apikey SET alias = `key` WHERE alias IS NULL"),
+        ("markdown", "updated_at", "ALTER TABLE markdown ADD COLUMN updated_at DATETIME NULL",
+         "UPDATE markdown SET updated_at = created_at WHERE updated_at IS NULL"),
+        ("markdown", "author", "ALTER TABLE markdown ADD COLUMN author VARCHAR(255) NULL",
+         "UPDATE markdown SET author = 'admin' WHERE author IS NULL"),
+        ("markdown", "last_modified_by", "ALTER TABLE markdown ADD COLUMN last_modified_by VARCHAR(255) NULL",
+         "UPDATE markdown SET last_modified_by = 'admin' WHERE last_modified_by IS NULL"),
+        ("markdown_patch", "author", "ALTER TABLE markdown_patch ADD COLUMN author VARCHAR(255) NULL",
+         "UPDATE markdown_patch SET author = 'admin' WHERE author IS NULL"),
+        ("markdown_patch", "last_modified_by", "ALTER TABLE markdown_patch ADD COLUMN last_modified_by VARCHAR(255) NULL",
+         "UPDATE markdown_patch SET last_modified_by = 'admin' WHERE last_modified_by IS NULL"),
+    ]
+    try:
+        with engine.begin() as conn:
+            for table, column, ddl, backfill in steps:
+                if not _column_exists(conn, table, column):
+                    conn.execute(text(ddl))
+                    if backfill:
+                        conn.execute(text(backfill))
+                    print(f"[ x ] migrated {table}.{column}")
+            # Unique constraint on apikey.alias once it is populated.
+            if not _index_exists(conn, "apikey", "uq_apikey_alias"):
+                conn.execute(text(
+                    "ALTER TABLE apikey ADD CONSTRAINT uq_apikey_alias UNIQUE (alias)"
+                ))
+                print("[ x ] migrated apikey.alias unique constraint")
+    except Exception as e:
+        # Don't block startup on a migration hiccup; surface loudly.
+        print(f"[ ! ] run_migrations error (continuing): {e}")
+
+
 def setup_db():
     if DB_SCHEMA_UPDATED:
         clear_db()
         print("[ x ] db cleared")
     create_all()
     print("[ x ] db created")
+    run_migrations()
+    print("[ x ] db migrations applied")
     run_scripts()
     print("[ x ] db scripts executed")
     init_payload()

db/migrations/2026-05-16_apikey_alias_authorship.sql

@@ -0,0 +1,79 @@
+-- ============================================================================
+-- Production migration: apikey.alias (unique) + markdown/patch authorship
+-- ============================================================================
+-- Idempotent. Safe to run multiple times. Target: MySQL 8 (no native
+-- ADD COLUMN IF NOT EXISTS, so columns are guarded via information_schema
+-- + a prepared statement). Mirrors the app's db.run_migrations(); running
+-- both is harmless.
+--
+-- Apply against the application schema, e.g.:
+--   docker exec -i mysql sh -c 'mysql -uroot -p"$MYSQL_ROOT_PASSWORD" hangmanlab' \
+--       < 2026-05-16_apikey_alias_authorship.sql
+-- ============================================================================
+
+SET @schema := DATABASE();
+
+-- ---- helper macro is not available in plain SQL; repeat the guarded block --
+
+-- apikey.alias --------------------------------------------------------------
+SET @c := (SELECT COUNT(*) FROM information_schema.columns
+           WHERE table_schema=@schema AND table_name='apikey' AND column_name='alias');
+SET @ddl := IF(@c=0,
+    'ALTER TABLE apikey ADD COLUMN alias VARCHAR(255) NULL',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+-- backfill: existing keys get alias = their (unique) key string
+UPDATE apikey SET alias = `key` WHERE alias IS NULL;
+
+-- apikey unique constraint on alias -----------------------------------------
+SET @i := (SELECT COUNT(*) FROM information_schema.statistics
+           WHERE table_schema=@schema AND table_name='apikey' AND index_name='uq_apikey_alias');
+SET @ddl := IF(@i=0,
+    'ALTER TABLE apikey ADD CONSTRAINT uq_apikey_alias UNIQUE (alias)',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+
+-- markdown.updated_at -------------------------------------------------------
+SET @c := (SELECT COUNT(*) FROM information_schema.columns
+           WHERE table_schema=@schema AND table_name='markdown' AND column_name='updated_at');
+SET @ddl := IF(@c=0,
+    'ALTER TABLE markdown ADD COLUMN updated_at DATETIME NULL',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+UPDATE markdown SET updated_at = created_at WHERE updated_at IS NULL;
+
+-- markdown.author -----------------------------------------------------------
+SET @c := (SELECT COUNT(*) FROM information_schema.columns
+           WHERE table_schema=@schema AND table_name='markdown' AND column_name='author');
+SET @ddl := IF(@c=0,
+    'ALTER TABLE markdown ADD COLUMN author VARCHAR(255) NULL',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+UPDATE markdown SET author = 'admin' WHERE author IS NULL;
+
+-- markdown.last_modified_by -------------------------------------------------
+SET @c := (SELECT COUNT(*) FROM information_schema.columns
+           WHERE table_schema=@schema AND table_name='markdown' AND column_name='last_modified_by');
+SET @ddl := IF(@c=0,
+    'ALTER TABLE markdown ADD COLUMN last_modified_by VARCHAR(255) NULL',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+UPDATE markdown SET last_modified_by = 'admin' WHERE last_modified_by IS NULL;
+
+-- markdown_patch.author -----------------------------------------------------
+SET @c := (SELECT COUNT(*) FROM information_schema.columns
+           WHERE table_schema=@schema AND table_name='markdown_patch' AND column_name='author');
+SET @ddl := IF(@c=0,
+    'ALTER TABLE markdown_patch ADD COLUMN author VARCHAR(255) NULL',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+UPDATE markdown_patch SET author = 'admin' WHERE author IS NULL;
+
+-- markdown_patch.last_modified_by -------------------------------------------
+SET @c := (SELECT COUNT(*) FROM information_schema.columns
+           WHERE table_schema=@schema AND table_name='markdown_patch' AND column_name='last_modified_by');
+SET @ddl := IF(@c=0,
+    'ALTER TABLE markdown_patch ADD COLUMN last_modified_by VARCHAR(255) NULL',
+    'DO 0');
+PREPARE st FROM @ddl; EXECUTE st; DEALLOCATE PREPARE st;
+UPDATE markdown_patch SET last_modified_by = 'admin' WHERE last_modified_by IS NULL;

db/models/APIKey.py

@@ -0,0 +1,29 @@
+from datetime import datetime, timedelta, UTC
+from sqlalchemy import Column, String, DateTime, Boolean, JSON
+from db.models import Base
+
+class APIKey(Base):
+    __tablename__ = 'apikey'
+
+    key = Column(String(64), primary_key=True)
+    # Stable human identity of the key. Unique; creating with an existing
+    # alias is treated as a renewal of that key (see api/apikey).
+    alias = Column(String(255), nullable=False, unique=True)
+    name = Column(String(255), nullable=False)
+    created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC))
+    last_used_at = Column(DateTime)
+    is_active = Column(Boolean, default=True)
+    roles = Column(JSON, nullable=False, default=list)
+    expire = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC) + timedelta(days=15))
+
+    def to_dict(self):
+        return {
+            "key": self.key,
+            "alias": self.alias,
+            "name": self.name,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "last_used_at": self.last_used_at.isoformat() if self.last_used_at else None,
+            "is_active": self.is_active,
+            "roles": self.roles,
+            "expire": self.expire.isoformat() if self.expire else None
+        } 

db/models/Markdown.py

@@ -9,7 +9,15 @@ class Markdown(Base):
     id = Column(Integer, primary_key=True)
     title = Column(String(255), nullable=False)
     content = Column(Text, nullable=False)
-    created_at = Column(DateTime, default=datetime.datetime.now(datetime.UTC))
+    created_at = Column(DateTime, default=lambda: datetime.datetime.now(datetime.UTC))
+    updated_at = Column(
+        DateTime,
+        default=lambda: datetime.datetime.now(datetime.UTC),
+        onupdate=lambda: datetime.datetime.now(datetime.UTC),
+    )
+    # Actor strings: alias of the API key, or 'admin' for KC-logged-in UI.
+    author = Column(String(255), nullable=True)
+    last_modified_by = Column(String(255), nullable=True)
     path_id = Column(Integer, ForeignKey('path.id'), nullable=False)
     order = Column(String(36), default=lambda: str(uuid.uuid4()))
     shortcut = Column(String(36), default="")
@@ -21,6 +29,9 @@ class Markdown(Base):
             'title': self.title,
             'content': self.content,
             'created_at': self.created_at,
+            'updated_at': self.updated_at,
+            'author': self.author,
+            'last_modified_by': self.last_modified_by,
             'path_id': self.path_id,
             'order': self.order,
             'shortcut': self.shortcut,

db/models/MarkdownPatch.py

@@ -0,0 +1,42 @@
+from sqlalchemy import Column, Text, Integer, String, DateTime, ForeignKey
+from db.models import Base
+import datetime
+
+
+class MarkdownPatch(Base):
+    """A markdown 'patch card' attached to a parent markdown document.
+
+    Its content is plain markdown text rendered below the parent's body.
+    Created/updated by admin|creator, deleted by admin; visibility is
+    inherited from the parent markdown (enforced in the API layer).
+    """
+    __tablename__ = 'markdown_patch'
+    id = Column(Integer, primary_key=True)
+    markdown_id = Column(
+        Integer, ForeignKey('markdown.id'), nullable=False, index=True
+    )
+    title = Column(String(255), nullable=True)
+    content = Column(Text, nullable=False)
+    # Actor strings: alias of the API key, or 'admin' for KC-logged-in UI.
+    author = Column(String(255), nullable=True)
+    last_modified_by = Column(String(255), nullable=True)
+    order = Column(Integer, default=0)
+    created_at = Column(DateTime, default=lambda: datetime.datetime.now(datetime.UTC))
+    updated_at = Column(
+        DateTime,
+        default=lambda: datetime.datetime.now(datetime.UTC),
+        onupdate=lambda: datetime.datetime.now(datetime.UTC),
+    )
+
+    def to_dict(self):
+        return {
+            'id': self.id,
+            'markdown_id': self.markdown_id,
+            'title': self.title,
+            'content': self.content,
+            'author': self.author,
+            'last_modified_by': self.last_modified_by,
+            'order': self.order,
+            'created_at': self.created_at,
+            'updated_at': self.updated_at,
+        }

db/models/Path.py

@@ -10,7 +10,7 @@ class Path(Base):
     name = Column(String(50), nullable=False)
     parent_id = Column(Integer, ForeignKey("path.id"), nullable=True)
     order = Column(String(36), default=lambda: str(uuid.uuid4()))
-    setting_id = Column(Integer, ForeignKey("path_settings.id"), nullable=True)
+    setting_id = Column(Integer, ForeignKey("path_setting.id"), nullable=True)
     __table_args__ = (UniqueConstraint("parent_id", "name", name="unique_parent_id_name"), )
     def to_dict(self):
         return {

db/models/PathSetting.py

@@ -4,7 +4,7 @@ from db.models import Base
 
 
 class PathSetting(Base):
-    __tablename__ = 'path_settings'
+    __tablename__ = 'path_setting'
     id = Column(Integer, primary_key=True, autoincrement=True)
     webhook_setting_id = Column(Integer, ForeignKey('webhook_setting.id'), nullable=True)
     template_setting_id = Column(Integer, ForeignKey('path_template.id'), nullable=True)

env_provider.py

@@ -23,6 +23,12 @@ KC_CLIENT_ID = os.getenv("KC_CLIENT_ID")
 FRONTEND_HOST = os.getenv("FRONTEND_HOST")
 BACKEND_HOST = os.getenv("BACKEND_HOST")
 
+def _redact(value):
+    if not value:
+        return "<unset>"
+    return f"<set:{len(str(value))} chars>"
+
+
 def summerize():
     return {
         "ENVIRONMENT": ENVIRONMENT,
@@ -30,9 +36,9 @@ def summerize():
         'DB_PORT': DB_PORT,
         'DB_NAME': DB_NAME,
         'DB_USER': DB_USER,
-        'DB_PASSWORD': DB_PASSWORD,
+        'DB_PASSWORD': _redact(DB_PASSWORD),
         'DB_SCHEMA_UPDATED': DB_SCHEMA_UPDATED,
-        'SESSION_SECRET_KEY': SESSION_SECRET_KEY,
+        'SESSION_SECRET_KEY': _redact(SESSION_SECRET_KEY),
         'KC_HOST': KC_HOST,
         'KC_REALM': KC_REALM,
         'KC_CLIENT_ID': KC_CLIENT_ID,

events/WebhookEventHandlers/__init__.py

@@ -1,14 +1,71 @@
 from sqlalchemy.orm import Session
 from db.models.Path import Path
+from db.models.PathSetting import PathSetting
 from db.models.Webhook import Webhook
 from db.models.WebhookSetting import WebhookSetting
+from events import MARKDOWN_CREATED_EVENT, MARKDOWN_UPDATED_EVENT, MARKDOWN_DELETED_EVENT, PATH_CREATED_EVENT, \
+    PATH_UPDATED_EVENT, PATH_DELETED_EVENT
 import abc
 import importlib
+import ipaddress
 import json
 import pkgutil
+import socket
 import requests
 import db
+from urllib.parse import urlsplit
 
+import logging
+logger = logging.getLogger(__name__)
+
+
+def is_safe_webhook_url(url):
+    """
+    Reject webhook targets that could be used for SSRF: only http/https,
+    and the resolved host must not be loopback / private / link-local /
+    reserved. Defends internal services and cloud metadata endpoints
+    (e.g. 169.254.169.254) even when the stored URL came from a backup.
+    """
+    try:
+        parts = urlsplit(url)
+    except Exception:
+        return False
+    if parts.scheme not in ("http", "https") or not parts.hostname:
+        return False
+    try:
+        infos = socket.getaddrinfo(parts.hostname, None)
+    except socket.gaierror:
+        return False
+    for info in infos:
+        ip = ipaddress.ip_address(info[4][0])
+        if (ip.is_private or ip.is_loopback or ip.is_link_local
+                or ip.is_reserved or ip.is_multicast or ip.is_unspecified):
+            return False
+    return True
+
+
+def parse_additional_headers(raw):
+    """Best-effort parse of the stored additional_header JSON object."""
+    if not raw:
+        return {}
+    try:
+        parsed = json.loads(raw)
+    except (ValueError, TypeError):
+        logger.warning("webhook additional_header is not valid JSON; ignoring")
+        return {}
+    if not isinstance(parsed, dict):
+        return {}
+    return {str(k): str(v) for k, v in parsed.items()}
+
+
+event_type_map = {
+    MARKDOWN_CREATED_EVENT: 'markdown_created_event',
+    MARKDOWN_UPDATED_EVENT: 'markdown_updated_event',
+    MARKDOWN_DELETED_EVENT: 'markdown_deleted_event',
+    PATH_CREATED_EVENT: 'path_created_event',
+    PATH_UPDATED_EVENT: 'path_updated_event',
+    PATH_DELETED_EVENT: 'path_deleted_event',
+}
 class WebhookEventHandler(abc.ABC):
     def __init__(self, event_type=0):
         self.event_type = event_type
@@ -17,21 +74,28 @@ class WebhookEventHandler(abc.ABC):
     def get_path_id(self, payload):
         pass
 
-    def __call__(self, **kwargs):
+    def __call__(self, *args, **kwargs):
         payload = kwargs['payload']
         path_id = self.get_path_id(payload)
         with db.get_db() as session:
             setting = self.get_setting(session, path_id)
             if setting is None:
                 return
-            headers = {'Content-Type': 'application/json'}
+            webhook_url = setting["webhook_url"]
-            if setting["additional_headers"] is not None:
+            if not is_safe_webhook_url(webhook_url):
-                headers.update(json.loads(setting["additional_headers"]))
+                logger.warning("blocked webhook to unsafe URL: %s", webhook_url)
+                return
+            headers = {'Content-Type': 'application/json', 'x-alchegos-event': event_type_map[self.event_type]}
+            headers.update(parse_additional_headers(setting.get("additional_header")))
+            body = json.dumps(payload, default=str)
             try:
-                response = requests.post(setting["webhook_url"], json=payload, headers=headers, timeout=5)
+                response = requests.post(
+                    webhook_url, data=body, headers=headers,
+                    timeout=5, allow_redirects=False,
+                )
                 response.raise_for_status()
-            except requests.RequestException as e:
+            except Exception as e:
-                print(e)
+                logger.warning("webhook delivery failed: %s", e)
 
     def get_setting(self, session: Session, path_id):
 
@@ -39,15 +103,22 @@ class WebhookEventHandler(abc.ABC):
         if path is None:
             return None
         p = path.to_dict()
-        webhook_setting = session.query(WebhookSetting).filter(WebhookSetting.path_id == path_id).first()
+        path_setting = session.query(PathSetting).get(path.setting_id)
+        if path_setting is None:
+            return None
+        webhook_setting = session.query(WebhookSetting).get(path_setting.webhook_setting_id)
 
         if webhook_setting is None and p["parent_id"] != 1:
             return self.get_setting(session, p["parent_id"])
         
+        # Check if webhook_setting is still None (e.g., when parent_id == 1 or no parent found)
+        if webhook_setting is None:
+            return None
+
         setting = webhook_setting.to_dict()
         if not setting["enabled"] or setting["on_events"] & self.event_type == 0:
             return None
-        webhook = session.query(Webhook).filter(Webhook.id == webhook_setting.webhook_id).first()
+        webhook = session.query(Webhook).get(webhook_setting.webhook_id)
         if webhook is None:
             return None
         setting["webhook_url"] = webhook.to_dict()["hook_url"]

req2.txt

@@ -1,9 +0,0 @@
-cryptography==44.0.0
-Flask==3.1.0
-Flask_Cors==5.0.0
-Flask_Limiter==3.9.2
-PyJWT==2.10.1
-PyJWT==2.10.1
-python-dotenv==1.0.1
-Requests==2.32.3
-SQLAlchemy==2.0.36

requirements-test.txt

@@ -0,0 +1,5 @@
+# Test dependencies
+pytest==7.4.0
+pytest-flask==1.2.0
+pytest-cov==4.1.0
+pytest-mock==3.11.1

requirements.txt

@@ -34,3 +34,5 @@ typing_extensions==4.12.2
 urllib3==2.2.3
 Werkzeug==3.1.3
 wrapt==1.17.0
+
+pytest~=8.3.5