Initial draft

This commit is contained in:
Kevin Veen-Birkenbach
2025-12-23 11:14:24 +01:00
parent 042f16f44b
commit 541eb04351
14 changed files with 445 additions and 0 deletions

View File

@@ -0,0 +1,7 @@
"""
matomo-bootstrap
Headless bootstrap tooling for Matomo:
- readiness checks
- admin/API token provisioning
"""
__all__ = []

View File

@@ -0,0 +1,23 @@
from .cli import parse_args
from .bootstrap import run_bootstrap
from .errors import BootstrapError
import sys
def main() -> int:
args = parse_args()
try:
token = run_bootstrap(args)
print(token)
return 0
except BootstrapError as exc:
print(f"[ERROR] {exc}", file=sys.stderr)
return 2
except Exception as exc:
print(f"[FATAL] {type(exc).__name__}: {exc}", file=sys.stderr)
return 3
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -0,0 +1,36 @@
import json
from .http import HttpClient
from .errors import TokenCreationError
def create_app_token(
client: HttpClient,
admin_user: str,
admin_password: str,
description: str,
) -> str:
status, body = client.post(
"/api.php",
{
"module": "API",
"method": "UsersManager.createAppSpecificTokenAuth",
"userLogin": admin_user,
"passwordConfirmation": admin_password,
"description": description,
"format": "json",
},
)
if status != 200:
raise TokenCreationError(f"HTTP {status} during token creation")
try:
data = json.loads(body)
except json.JSONDecodeError as exc:
raise TokenCreationError("Invalid JSON from Matomo API") from exc
token = data.get("value")
if not token:
raise TokenCreationError(f"Unexpected response: {data}")
return token

View File

@@ -0,0 +1,35 @@
from argparse import Namespace
from .health import assert_matomo_ready
from .http import HttpClient
from .api_tokens import create_app_token
from .install.web_installer import ensure_installed
def run_bootstrap(args: Namespace) -> str:
# 1. Matomo erreichbar?
assert_matomo_ready(args.base_url, timeout=args.timeout)
# 2. Installation sicherstellen (NO-OP wenn bereits installiert)
ensure_installed(
base_url=args.base_url,
admin_user=args.admin_user,
admin_password=args.admin_password,
admin_email=args.admin_email,
debug=args.debug,
)
# 3. API-Token erzeugen
client = HttpClient(
base_url=args.base_url,
timeout=args.timeout,
debug=args.debug,
)
token = create_app_token(
client=client,
admin_user=args.admin_user,
admin_password=args.admin_password,
description=args.token_description,
)
return token

View File

@@ -0,0 +1,17 @@
import argparse
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(
description="Headless bootstrap tool for Matomo (installation + API token provisioning)"
)
p.add_argument("--base-url", required=True, help="Matomo base URL")
p.add_argument("--admin-user", required=True)
p.add_argument("--admin-password", required=True)
p.add_argument("--admin-email", required=True)
p.add_argument("--token-description", default="matomo-bootstrap")
p.add_argument("--timeout", type=int, default=20)
p.add_argument("--debug", action="store_true")
return p.parse_args()

View File

@@ -0,0 +1,10 @@
class BootstrapError(RuntimeError):
"""Base error for matomo-bootstrap."""
class MatomoNotReadyError(BootstrapError):
"""Matomo is not reachable or not initialized."""
class TokenCreationError(BootstrapError):
"""Failed to create API token."""

View File

@@ -0,0 +1,13 @@
import urllib.request
from .errors import MatomoNotReadyError
def assert_matomo_ready(base_url: str, timeout: int = 10) -> None:
try:
with urllib.request.urlopen(base_url, timeout=timeout) as resp:
html = resp.read().decode("utf-8", errors="replace")
except Exception as exc:
raise MatomoNotReadyError(f"Matomo not reachable: {exc}") from exc
if "Matomo" not in html and "piwik" not in html.lower():
raise MatomoNotReadyError("Matomo UI not detected at base URL")

View File

@@ -0,0 +1,28 @@
import urllib.request
import urllib.parse
import http.cookiejar
from typing import Dict, Tuple
class HttpClient:
def __init__(self, base_url: str, timeout: int = 20, debug: bool = False):
self.base_url = base_url.rstrip("/")
self.timeout = timeout
self.debug = debug
self.cookies = http.cookiejar.CookieJar()
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.cookies)
)
def post(self, path: str, data: Dict[str, str]) -> Tuple[int, str]:
url = self.base_url + path
encoded = urllib.parse.urlencode(data).encode()
if self.debug:
print(f"[HTTP] POST {url} keys={list(data.keys())}")
req = urllib.request.Request(url, data=encoded, method="POST")
with self.opener.open(req, timeout=self.timeout) as resp:
body = resp.read().decode("utf-8", errors="replace")
return resp.status, body

View File

View File

@@ -0,0 +1,160 @@
import os
import sys
import time
import urllib.request
MATOMO_URL = os.environ.get("MATOMO_URL", "http://127.0.0.1:8080")
ADMIN_USER = os.environ.get("MATOMO_ADMIN_USER", "administrator")
ADMIN_PASSWORD = os.environ.get("MATOMO_ADMIN_PASSWORD", "AdminSecret123!")
ADMIN_EMAIL = os.environ.get("MATOMO_ADMIN_EMAIL", "admin@example.org")
DB_HOST = os.environ.get("MATOMO_DB_HOST", "db")
DB_USER = os.environ.get("MATOMO_DB_USER", "matomo")
DB_PASS = os.environ.get("MATOMO_DB_PASS", "matomo_pw")
DB_NAME = os.environ.get("MATOMO_DB_NAME", "matomo")
DB_PREFIX = os.environ.get("MATOMO_DB_PREFIX", "matomo_")
def wait_http(url: str, timeout: int = 180) -> None:
print(f"[install] Waiting for Matomo HTTP at {url} ...")
for i in range(timeout):
try:
with urllib.request.urlopen(url, timeout=2) as resp:
_ = resp.read(1024)
print("[install] Matomo HTTP reachable.")
return
except Exception:
if i % 5 == 0:
print(f"[install] still waiting ({i}/{timeout}) …")
time.sleep(1)
raise RuntimeError(f"Matomo did not become reachable after {timeout}s: {url}")
def is_installed(url: str) -> bool:
try:
with urllib.request.urlopen(url, timeout=3) as resp:
html = resp.read().decode(errors="ignore").lower()
return ("module=login" in html) or ("matomo login" in html)
except Exception:
return False
def ensure_installed(
base_url: str,
admin_user: str,
admin_password: str,
admin_email: str,
debug: bool = False,
) -> None:
"""
Ensure Matomo is installed.
NO-OP if already installed.
"""
# Propagate config to installer via ENV (single source of truth)
os.environ["MATOMO_URL"] = base_url
os.environ["MATOMO_ADMIN_USER"] = admin_user
os.environ["MATOMO_ADMIN_PASSWORD"] = admin_password
os.environ["MATOMO_ADMIN_EMAIL"] = admin_email
rc = main()
if rc != 0:
raise RuntimeError("Matomo installation failed")
def main() -> int:
wait_http(MATOMO_URL)
if is_installed(MATOMO_URL):
print("[install] Matomo already installed. Skipping installer.")
return 0
try:
from playwright.sync_api import sync_playwright
except Exception as exc:
print("[install] Playwright not available.", file=sys.stderr)
print(
"Install with: python3 -m pip install playwright && python3 -m playwright install chromium",
file=sys.stderr,
)
print(f"Reason: {exc}", file=sys.stderr)
return 2
print("[install] Running Matomo web installer via headless browser...")
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto(MATOMO_URL, wait_until="domcontentloaded")
def click_next():
for label in ["Next", "Continue", "Start Installation", "Proceed"]:
btn = page.get_by_role("button", name=label)
if btn.count() > 0:
btn.first.click()
return True
return False
# Welcome / system check
page.wait_for_timeout(500)
click_next()
page.wait_for_timeout(500)
click_next()
# Database setup
page.wait_for_timeout(500)
page.get_by_label("Database Server").fill(DB_HOST)
page.get_by_label("Login").fill(DB_USER)
page.get_by_label("Password").fill(DB_PASS)
page.get_by_label("Database Name").fill(DB_NAME)
try:
page.get_by_label("Tables Prefix").fill(DB_PREFIX)
except Exception:
pass
click_next()
# Tables creation
page.wait_for_timeout(500)
click_next()
# Super user
page.wait_for_timeout(500)
page.get_by_label("Login").fill(ADMIN_USER)
page.get_by_label("Password").fill(ADMIN_PASSWORD)
try:
page.get_by_label("Password (repeat)").fill(ADMIN_PASSWORD)
except Exception:
pass
page.get_by_label("Email").fill(ADMIN_EMAIL)
click_next()
# First website
page.wait_for_timeout(500)
try:
page.get_by_label("Name").fill("Bootstrap Site")
except Exception:
pass
try:
page.get_by_label("URL").fill("http://example.invalid")
except Exception:
pass
click_next()
# Finish
page.wait_for_timeout(500)
click_next()
browser.close()
time.sleep(2)
if not is_installed(MATOMO_URL):
print("[install] Installer did not reach installed state.", file=sys.stderr)
return 3
print("[install] Installation finished.")
return 0
if __name__ == "__main__":
raise SystemExit(main())