Magma Calculator Service Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: Build a containerized REST API that executes Magma code in an nsjail sandbox and returns parsed results as JSON.

Architecture: Single FastAPI app behind uvicorn with TLS. Each request spawns nsjail -> magma -w -n as a subprocess. In-memory IP rate limiting. CORS for browser access from GitHub Pages frontend.

Tech Stack: Python 3.11, FastAPI, uvicorn, pydantic-settings, pytest, httpx (test client), Docker, nsjail

Design document: /home/edgarcosta/magma/svn/docs/plans/2026-02-04-magma-calculator-design.md

Task 1: Project scaffolding and config

Files:

Create: requirements.txt
Create: app/__init__.py
Create: app/config.py
Create: tests/__init__.py
Create: tests/test_config.py
Create: calculator.env.example
Create: .gitignore

Step 1: Create .gitignore

__pycache__/
*.pyc
.pytest_cache/
*.egg-info/
.env
venv/
.venv/

Step 2: Create requirements.txt

fastapi>=0.104.0
uvicorn[standard]>=0.24.0
pydantic-settings>=2.1.0
pytest>=7.4.0
httpx>=0.25.0

Step 3: Install dependencies

Run: cd /home/edgarcosta/magma/calculator && python3 -m venv .venv && . .venv/bin/activate && pip install -r requirements.txt

Step 4: Create empty init.py files

Create empty app/__init__.py and tests/__init__.py.

Step 5: Write the failing config test

# tests/test_config.py
from app.config import Settings


def test_default_settings():
    settings = Settings()
    assert settings.magma_timeout == 120
    assert settings.magma_cpu_timeout == 120
    assert settings.magma_memory_mb == 400
    assert settings.magma_input_kb == 50
    assert settings.magma_output_kb == 20
    assert settings.max_concurrent == 4
    assert settings.port == 8080
    assert settings.rate_limit_per_minute == 30
    assert settings.rate_limit_per_hour == 200
    assert settings.allowed_origin == "https://magma-maths.org,http://localhost"
    assert settings.turnstile_enabled is False
    assert settings.turnstile_secret_key == ""
    assert settings.tls_cert_file == "/certs/live/calc.magma-maths.org/fullchain.pem"
    assert settings.tls_key_file == "/certs/live/calc.magma-maths.org/privkey.pem"


def test_settings_from_env(monkeypatch):
    monkeypatch.setenv("MAGMA_TIMEOUT", "300")
    monkeypatch.setenv("MAGMA_MEMORY_MB", "800")
    monkeypatch.setenv("ALLOWED_ORIGIN", "https://example.com")
    settings = Settings()
    assert settings.magma_timeout == 300
    assert settings.magma_memory_mb == 800
    assert settings.allowed_origin == "https://example.com"


def test_allowed_origins_list():
    settings = Settings()
    origins = settings.allowed_origins_list
    assert "https://magma-maths.org" in origins
    assert "http://localhost" in origins

Step 6: Run test to verify it fails

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_config.py -v Expected: FAIL (ImportError, module not found)

Step 7: Write config implementation

# app/config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Magma execution
    magma_timeout: int = 120
    magma_cpu_timeout: int = 120
    magma_memory_mb: int = 400
    magma_input_kb: int = 50
    magma_output_kb: int = 20

    # Service
    max_concurrent: int = 4
    port: int = 8080

    # TLS
    tls_cert_file: str = "/certs/live/calc.magma-maths.org/fullchain.pem"
    tls_key_file: str = "/certs/live/calc.magma-maths.org/privkey.pem"

    # Rate limiting
    rate_limit_per_minute: int = 30
    rate_limit_per_hour: int = 200

    # CORS
    allowed_origin: str = "https://magma-maths.org,http://localhost"

    # Optional Turnstile
    turnstile_enabled: bool = False
    turnstile_secret_key: str = ""

    @property
    def allowed_origins_list(self) -> list[str]:
        return [o.strip() for o in self.allowed_origin.split(",")]

    @property
    def magma_input_bytes(self) -> int:
        return self.magma_input_kb * 1024

    @property
    def magma_output_bytes(self) -> int:
        return self.magma_output_kb * 1024

Step 8: Run tests to verify they pass

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_config.py -v Expected: 3 tests PASS

Step 9: Create calculator.env.example

# Magma execution
MAGMA_TIMEOUT=120
MAGMA_CPU_TIMEOUT=120
MAGMA_MEMORY_MB=400
MAGMA_INPUT_KB=50
MAGMA_OUTPUT_KB=20

# Service
MAX_CONCURRENT=4
PORT=8080

# TLS
TLS_CERT_FILE=/certs/live/calc.magma-maths.org/fullchain.pem
TLS_KEY_FILE=/certs/live/calc.magma-maths.org/privkey.pem

# Rate limiting
RATE_LIMIT_PER_MINUTE=30
RATE_LIMIT_PER_HOUR=200

# CORS
ALLOWED_ORIGIN=https://magma-maths.org,http://localhost

# Optional Turnstile
TURNSTILE_ENABLED=false
TURNSTILE_SECRET_KEY=

Step 10: Commit

git add .gitignore requirements.txt calculator.env.example app/__init__.py app/config.py tests/__init__.py tests/test_config.py
git commit -m "feat: project scaffolding and config with env var support"

Task 2: Magma output parser

This is the core parsing logic. It takes raw Magma stdout (which includes banner, body, and footer) and extracts structured data. This can be fully unit-tested without Magma installed.

Files:

Create: app/parser.py
Create: tests/test_parser.py

Step 1: Write failing parser tests

These tests use realistic Magma output samples. The parser operates on the complete stdout string (not streaming).

# tests/test_parser.py
from app.parser import parse_magma_output


SAMPLE_BANNER = "Magma V2.29-4     Fri Jan 31 2026 12:00:00 on linux   [Seed = 1234567890]\n"
SAMPLE_QUIT = "quit.\n"
SAMPLE_BODY = "2\n"
SAMPLE_FOOTER = "Total time: 0.050 seconds, Total memory usage: 12.34MB\n"


def test_parse_simple_output():
    """Parse a simple 'print 1+1;' output."""
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + SAMPLE_BODY + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert result.version == "2.29-4"
    assert result.seed == 1234567890
    assert result.stdout == "2\n"
    assert result.time_sec == 0.05
    assert result.memory == "12.34MB"
    assert result.truncated is False
    assert result.warnings == []


def test_parse_extracts_version_variants():
    """Version can be like V2.29-4 or V2.28."""
    stdout = "Magma V2.28     Fri Jan 31 2026 [Seed = 999]\nquit.\nok\nTotal time: 1.000 seconds, Total memory usage: 5.00MB\n"
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert result.version == "2.28"
    assert result.seed == 999


def test_parse_multiline_body():
    """Body can be multiple lines."""
    body = "line1\nline2\nline3\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert result.stdout == body


def test_parse_strips_machine_type():
    """Machine type line should be removed from body."""
    body = "Machine type: X86_64-linux\nresult\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert "Machine type" not in result.stdout
    assert "result\n" in result.stdout


def test_parse_truncates_long_output():
    """Output exceeding max_output_bytes is truncated."""
    body = "x" * 100 + "\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=50)
    assert len(result.stdout) == 50
    assert result.truncated is True
    assert "The output is too long and has been truncated." in result.warnings


def test_parse_detects_memory_limit():
    """Memory limit exceeded adds warning."""
    body = "User memory limit exceeded\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert any("memory limit" in w for w in result.warnings)


def test_parse_detects_runtime_error():
    """Runtime errors add warning."""
    body = "Runtime error in 'foo': something\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert any("error occurred" in w.lower() for w in result.warnings)


def test_parse_detects_user_error():
    """User errors add warning."""
    body = "User error: bad input\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert any("error occurred" in w.lower() for w in result.warnings)


def test_parse_detects_internal_error():
    """Internal errors add warning."""
    body = "Something (internal error)\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert any("error occurred" in w.lower() for w in result.warnings)


def test_parse_detects_illegal_syscall():
    """Illegal system call adds warning."""
    body = "Illegal system call\n"
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + body + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert any("error occurred" in w.lower() for w in result.warnings)


def test_parse_empty_body():
    """Empty output between quit and footer."""
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + SAMPLE_FOOTER
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert result.stdout == ""


def test_parse_no_footer():
    """If Magma is killed before footer, time/memory are None."""
    stdout = SAMPLE_BANNER + SAMPLE_QUIT + "partial output\n"
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert result.stdout == "partial output\n"
    assert result.time_sec is None
    assert result.memory is None


def test_parse_no_banner():
    """If Magma crashes before banner, version/seed are None."""
    stdout = ""
    result = parse_magma_output(stdout, max_output_bytes=20480)
    assert result.version is None
    assert result.seed is None
    assert result.stdout == ""

Step 2: Run tests to verify they fail

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_parser.py -v Expected: FAIL (ImportError)

Step 3: Write parser implementation

# app/parser.py
import re
from dataclasses import dataclass, field


@dataclass
class ParseResult:
    version: str | None = None
    seed: int | None = None
    stdout: str = ""
    time_sec: float | None = None
    memory: str | None = None
    truncated: bool = False
    warnings: list[str] = field(default_factory=list)


_RE_VERSION = re.compile(r"Magma V(\d+\.\d+(-[A-Z]*\d+)?)")
_RE_SEED = re.compile(r"\[Seed = (\d+)\]")
_RE_FOOTER_START = re.compile(r"Total time:\s+\d+\.\d+ seconds, Total memory usage: ")
_RE_TIME = re.compile(r"Total time:\s+(\d+\.\d+)")
_RE_MEMORY = re.compile(r"Total memory usage: (\d+\.\d+[A-Z]+)")
_RE_MACHINE_TYPE = re.compile(r"Machine type: .*\n")

_ERROR_PATTERNS = [
    "User error: ",
    "Runtime error in ",
    "(internal error)",
    "Illegal system call",
]


def parse_magma_output(stdout: str, max_output_bytes: int) -> ParseResult:
    result = ParseResult()

    if not stdout:
        return result

    # Phase 1: Banner - everything before "quit.\n"
    quit_idx = stdout.find("quit.\n")
    if quit_idx == -1:
        # No quit marker - Magma may have crashed before processing input
        _extract_banner(stdout, result)
        return result

    banner = stdout[:quit_idx]
    rest = stdout[quit_idx + len("quit.\n"):]

    _extract_banner(banner, result)

    # Phase 2/3: Split body from footer
    footer_match = _RE_FOOTER_START.search(rest)
    if footer_match:
        body = rest[:footer_match.start()]
        footer = rest[footer_match.start():]
        _extract_footer(footer, result)
    else:
        body = rest

    # Process body
    body = body.rstrip("\n")
    if body:
        body += "\n"
    else:
        body = ""

    # Strip Machine type lines
    body = _RE_MACHINE_TYPE.sub("", body)

    # Detect memory limit
    if "User memory limit" in body:
        result.warnings.append(
            "The computation exceeded the memory limit and so was terminated prematurely."
        )

    # Detect errors (only add warning once)
    for pattern in _ERROR_PATTERNS:
        if pattern in body:
            result.warnings.append("An error occurred. See the output for details.")
            break

    # Truncate if needed
    if len(body) > max_output_bytes:
        body = body[:max_output_bytes]
        result.truncated = True
        result.warnings.append("The output is too long and has been truncated.")

    result.stdout = body
    return result


def _extract_banner(text: str, result: ParseResult) -> None:
    m = _RE_VERSION.search(text)
    if m:
        result.version = m.group(1)
    m = _RE_SEED.search(text)
    if m:
        result.seed = int(m.group(1))


def _extract_footer(text: str, result: ParseResult) -> None:
    m = _RE_TIME.search(text)
    if m:
        result.time_sec = float(m.group(1))
    m = _RE_MEMORY.search(text)
    if m:
        result.memory = m.group(1)

Step 4: Run tests to verify they pass

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_parser.py -v Expected: All PASS

Step 5: Commit

git add app/parser.py tests/test_parser.py
git commit -m "feat: magma output parser with banner/body/footer extraction"

Task 3: Stderr parser

Separate function that processes stderr into warnings.

Files:

Modify: app/parser.py
Modify: tests/test_parser.py

Step 1: Add failing stderr tests

Append to tests/test_parser.py:

from app.parser import parse_stderr_warnings


def test_stderr_alarm_clock():
    warnings = parse_stderr_warnings("Alarm clock\n")
    assert len(warnings) == 1
    assert "time limit" in warnings[0]


def test_stderr_cputime_limit():
    warnings = parse_stderr_warnings("Cputime limit exceeded\n")
    assert len(warnings) == 1
    assert "time limit" in warnings[0]


def test_stderr_killed():
    warnings = parse_stderr_warnings("Killed\n")
    assert len(warnings) == 1
    assert "time limit" in warnings[0]


def test_stderr_fatal_error():
    warnings = parse_stderr_warnings("Magma: Fatal Error\n")
    assert len(warnings) == 1
    assert "fatal error" in warnings[0].lower()


def test_stderr_empty():
    warnings = parse_stderr_warnings("")
    assert warnings == []


def test_stderr_none():
    warnings = parse_stderr_warnings(None)
    assert warnings == []


def test_stderr_unknown():
    """Unknown stderr content is ignored (not passed to user)."""
    warnings = parse_stderr_warnings("some random debug output\n")
    assert warnings == []

Step 2: Run tests to verify they fail

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_parser.py -v -k stderr Expected: FAIL (ImportError)

Step 3: Add stderr parsing to parser.py

Append to app/parser.py:

def parse_stderr_warnings(stderr: str | None) -> list[str]:
    if not stderr:
        return []
    warnings = []
    if "Alarm clock" in stderr or "Cputime limit exceeded" in stderr or "Killed" in stderr:
        warnings.append(
            "The computation exceeded the time limit and so was terminated prematurely."
        )
    if "Magma: Fatal Error" in stderr:
        warnings.append("A fatal error occurred and Magma was forced to exit.")
    return warnings

Step 4: Run tests to verify they pass

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_parser.py -v Expected: All PASS

Step 5: Commit

git add app/parser.py tests/test_parser.py
git commit -m "feat: stderr warning parser for timeout/crash detection"

Task 4: Rate limiter

In-memory IP-based rate limiting.

Files:

Create: app/ratelimit.py
Create: tests/test_ratelimit.py

Step 1: Write failing rate limiter tests

# tests/test_ratelimit.py
import time
from app.ratelimit import RateLimiter


def test_allows_first_request():
    limiter = RateLimiter(per_minute=5, per_hour=100)
    assert limiter.is_allowed("1.2.3.4") is True


def test_blocks_after_per_minute_limit():
    limiter = RateLimiter(per_minute=3, per_hour=100)
    for _ in range(3):
        assert limiter.is_allowed("1.2.3.4") is True
    assert limiter.is_allowed("1.2.3.4") is False


def test_different_ips_independent():
    limiter = RateLimiter(per_minute=2, per_hour=100)
    assert limiter.is_allowed("1.1.1.1") is True
    assert limiter.is_allowed("1.1.1.1") is True
    assert limiter.is_allowed("1.1.1.1") is False
    # Different IP is still allowed
    assert limiter.is_allowed("2.2.2.2") is True


def test_blocks_after_per_hour_limit():
    limiter = RateLimiter(per_minute=1000, per_hour=5)
    for _ in range(5):
        assert limiter.is_allowed("1.2.3.4") is True
    assert limiter.is_allowed("1.2.3.4") is False


def test_cleanup_removes_old_entries():
    limiter = RateLimiter(per_minute=1000, per_hour=1000)
    limiter.is_allowed("1.2.3.4")
    assert "1.2.3.4" in limiter._requests
    # Manually age the entry
    limiter._requests["1.2.3.4"] = [time.time() - 7200]
    limiter.cleanup()
    assert "1.2.3.4" not in limiter._requests

Step 2: Run tests to verify they fail

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_ratelimit.py -v Expected: FAIL (ImportError)

Step 3: Write rate limiter implementation

# app/ratelimit.py
import time


class RateLimiter:
    def __init__(self, per_minute: int, per_hour: int):
        self.per_minute = per_minute
        self.per_hour = per_hour
        self._requests: dict[str, list[float]] = {}

    def is_allowed(self, ip: str) -> bool:
        now = time.time()
        if ip not in self._requests:
            self._requests[ip] = []

        timestamps = self._requests[ip]

        # Count requests in last minute
        one_minute_ago = now - 60
        recent_minute = sum(1 for t in timestamps if t > one_minute_ago)
        if recent_minute >= self.per_minute:
            return False

        # Count requests in last hour
        one_hour_ago = now - 3600
        recent_hour = sum(1 for t in timestamps if t > one_hour_ago)
        if recent_hour >= self.per_hour:
            return False

        timestamps.append(now)
        return True

    def cleanup(self) -> None:
        """Remove entries older than 1 hour."""
        cutoff = time.time() - 3600
        to_delete = []
        for ip, timestamps in self._requests.items():
            self._requests[ip] = [t for t in timestamps if t > cutoff]
            if not self._requests[ip]:
                to_delete.append(ip)
        for ip in to_delete:
            del self._requests[ip]

Step 4: Run tests to verify they pass

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_ratelimit.py -v Expected: All PASS

Step 5: Commit

git add app/ratelimit.py tests/test_ratelimit.py
git commit -m "feat: in-memory IP-based rate limiter"

Task 5: Executor (subprocess management)

The executor wraps code, spawns nsjail+magma, collects output, enforces timeouts. This module is hard to unit test without magma/nsjail, so we test the code-wrapping logic and mock the subprocess for the rest.

Files:

Create: app/executor.py
Create: tests/test_executor.py

Step 1: Write failing executor tests

# tests/test_executor.py
from app.executor import wrap_magma_code, ExecutionResult
from app.config import Settings


def test_wrap_magma_code():
    settings = Settings()
    wrapped = wrap_magma_code("print 1+1;", settings.magma_timeout)
    assert "Alarm(119);" in wrapped
    assert "SetIgnorePrompt(true);" in wrapped
    assert "print 1+1;" in wrapped
    assert wrapped.endswith(";\nquit;\n")


def test_wrap_magma_code_custom_timeout():
    wrapped = wrap_magma_code("x := 5;", 300)
    assert "Alarm(299);" in wrapped


def test_execution_result_dataclass():
    result = ExecutionResult(
        stdout="output",
        stderr="",
        exit_code=0,
    )
    assert result.stdout == "output"
    assert result.exit_code == 0

Step 2: Run tests to verify they fail

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_executor.py -v Expected: FAIL (ImportError)

Step 3: Write executor implementation

# app/executor.py
import asyncio
import os
import tempfile
from dataclasses import dataclass

from app.config import Settings


@dataclass
class ExecutionResult:
    stdout: str
    stderr: str
    exit_code: int


def wrap_magma_code(code: str, timeout: int) -> str:
    alarm_timeout = timeout - 1
    return (
        f"Alarm({alarm_timeout});\n"
        f"SetIgnorePrompt(true);\n"
        f"{code}\n"
        f";\n"
        f"quit;\n"
    )


async def execute_magma(code: str, settings: Settings) -> ExecutionResult:
    wrapped = wrap_magma_code(code, settings.magma_timeout)

    # Write wrapped code to temp file
    fd, tmppath = tempfile.mkstemp(suffix=".m", prefix="magma_")
    try:
        with os.fdopen(fd, "w") as f:
            f.write(wrapped)

        cmd = [
            "nsjail",
            "--config", "/app/nsjail.cfg",
            "--", "magma", "-w", "-n", tmppath,
        ]

        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        try:
            stdout_bytes, stderr_bytes = await asyncio.wait_for(
                proc.communicate(),
                timeout=settings.magma_timeout + 2,
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            return ExecutionResult(
                stdout="",
                stderr="Killed",
                exit_code=-1,
            )

        return ExecutionResult(
            stdout=stdout_bytes.decode("utf-8", errors="replace"),
            stderr=stderr_bytes.decode("utf-8", errors="replace"),
            exit_code=proc.returncode or 0,
        )
    finally:
        try:
            os.unlink(tmppath)
        except OSError:
            pass

Step 4: Run tests to verify they pass

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_executor.py -v Expected: All PASS

Step 5: Commit

git add app/executor.py tests/test_executor.py
git commit -m "feat: executor with code wrapping and nsjail subprocess management"

Task 6: FastAPI application and endpoints

The main application tying everything together: /execute, /health, CORS, rate limiting, concurrency control.

Files:

Create: app/main.py
Create: tests/test_main.py

Step 1: Write failing endpoint tests

These tests use FastAPI's test client and mock the executor (no real Magma needed).

# tests/test_main.py
import pytest
from unittest.mock import patch, AsyncMock
from fastapi.testclient import TestClient

from app.executor import ExecutionResult


@pytest.fixture
def client():
    from app.main import app
    return TestClient(app)


def test_health(client):
    resp = client.get("/health")
    assert resp.status_code == 200
    assert resp.json() == {"status": "ok"}


MOCK_MAGMA_STDOUT = (
    "Magma V2.29-4     Fri Jan 31 2026 [Seed = 42]\n"
    "quit.\n"
    "2\n"
    "Total time: 0.050 seconds, Total memory usage: 12.34MB\n"
)


@patch("app.main.execute_magma", new_callable=AsyncMock)
def test_execute_success(mock_exec, client):
    mock_exec.return_value = ExecutionResult(
        stdout=MOCK_MAGMA_STDOUT,
        stderr="",
        exit_code=0,
    )
    resp = client.post("/execute", json={"code": "print 1+1;"})
    assert resp.status_code == 200
    data = resp.json()
    assert data["success"] is True
    assert data["stdout"] == "2\n"
    assert data["magma"]["version"] == "2.29-4"
    assert data["magma"]["seed"] == 42
    assert data["magma"]["time_sec"] == 0.05
    assert data["magma"]["memory"] == "12.34MB"
    assert data["truncated"] is False
    assert data["warnings"] == []


def test_execute_missing_code(client):
    resp = client.post("/execute", json={})
    assert resp.status_code == 422


def test_execute_input_too_large(client):
    big_code = "x" * (50 * 1024 + 1)
    resp = client.post("/execute", json={"code": big_code})
    assert resp.status_code == 413


@patch("app.main.execute_magma", new_callable=AsyncMock)
def test_execute_timeout(mock_exec, client):
    mock_exec.return_value = ExecutionResult(
        stdout="Magma V2.29-4 [Seed = 1]\nquit.\n",
        stderr="Alarm clock\n",
        exit_code=0,
    )
    resp = client.post("/execute", json={"code": "while true do end while;"})
    assert resp.status_code == 200
    data = resp.json()
    assert any("time limit" in w for w in data["warnings"])


def test_cors_preflight(client):
    resp = client.options(
        "/execute",
        headers={
            "Origin": "https://magma-maths.org",
            "Access-Control-Request-Method": "POST",
        },
    )
    assert resp.status_code == 200
    assert "https://magma-maths.org" in resp.headers.get(
        "access-control-allow-origin", ""
    )


def test_cors_localhost_any_port(client):
    resp = client.options(
        "/execute",
        headers={
            "Origin": "http://localhost:5000",
            "Access-Control-Request-Method": "POST",
        },
    )
    assert resp.status_code == 200
    assert "http://localhost:5000" in resp.headers.get(
        "access-control-allow-origin", ""
    )

Step 2: Run tests to verify they fail

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_main.py -v Expected: FAIL (ImportError)

Step 3: Write main.py

# app/main.py
import asyncio
import logging
import json
import re
import time

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from app.config import Settings
from app.executor import execute_magma, ExecutionResult
from app.parser import parse_magma_output, parse_stderr_warnings
from app.ratelimit import RateLimiter

settings = Settings()
rate_limiter = RateLimiter(
    per_minute=settings.rate_limit_per_minute,
    per_hour=settings.rate_limit_per_hour,
)
semaphore = asyncio.Semaphore(settings.max_concurrent)

logger = logging.getLogger("calculator")
logging.basicConfig(level=logging.INFO, format="%(message)s")

app = FastAPI(docs_url=None, redoc_url=None)


# Custom CORS origin check: allow http://localhost with any port
_localhost_origins = [
    o for o in settings.allowed_origins_list if o == "http://localhost"
]
_fixed_origins = [
    o for o in settings.allowed_origins_list if o != "http://localhost"
]


def _origin_allowed(origin: str) -> bool:
    if origin in _fixed_origins:
        return True
    if _localhost_origins and re.match(r"^http://localhost(:\d+)?$", origin):
        return True
    return False


@app.middleware("http")
async def cors_middleware(request: Request, call_next):
    origin = request.headers.get("origin", "")

    if request.method == "OPTIONS":
        if _origin_allowed(origin):
            return JSONResponse(
                status_code=200,
                headers={
                    "Access-Control-Allow-Origin": origin,
                    "Access-Control-Allow-Methods": "POST, GET, OPTIONS",
                    "Access-Control-Allow-Headers": "*",
                    "Access-Control-Max-Age": "3600",
                },
            )
        return JSONResponse(status_code=403)

    response = await call_next(request)

    if origin and _origin_allowed(origin):
        response.headers["Access-Control-Allow-Origin"] = origin

    return response


class ExecuteRequest(BaseModel):
    code: str


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.post("/execute")
async def execute(req: ExecuteRequest, request: Request):
    start_time = time.time()
    client_ip = request.client.host if request.client else "unknown"

    # Check input size
    if len(req.code.encode("utf-8")) > settings.magma_input_bytes:
        return JSONResponse(
            status_code=413,
            content={"error": "Input too large"},
        )

    # Check rate limit
    if not rate_limiter.is_allowed(client_ip):
        return JSONResponse(
            status_code=429,
            content={"error": "Rate limit exceeded"},
            headers={"Retry-After": "60"},
        )

    # Acquire concurrency slot
    if semaphore.locked() and semaphore._value == 0:
        return JSONResponse(
            status_code=503,
            content={"error": "All execution slots busy"},
        )

    async with semaphore:
        result: ExecutionResult = await execute_magma(req.code, settings)

    # Parse output
    parsed = parse_magma_output(result.stdout, settings.magma_output_bytes)
    stderr_warnings = parse_stderr_warnings(result.stderr)
    all_warnings = parsed.warnings + stderr_warnings

    success = result.exit_code == 0 and not stderr_warnings

    response_data = {
        "success": success,
        "stdout": parsed.stdout,
        "exit_code": result.exit_code,
        "truncated": parsed.truncated,
        "magma": {
            "version": parsed.version,
            "seed": parsed.seed,
            "time_sec": parsed.time_sec,
            "memory": parsed.memory,
        },
        "warnings": all_warnings,
    }

    if not success and stderr_warnings:
        response_data["error"] = stderr_warnings[0]

    elapsed = time.time() - start_time
    logger.info(
        json.dumps({
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "client_ip": client_ip,
            "input_size": len(req.code),
            "elapsed_sec": round(elapsed, 3),
            "success": success,
            "warnings": all_warnings,
        })
    )

    return response_data


if __name__ == "__main__":
    import os
    import uvicorn

    ssl_kwargs = {}
    cert = settings.tls_cert_file
    key = settings.tls_key_file
    if os.path.exists(cert) and os.path.exists(key):
        ssl_kwargs["ssl_certfile"] = cert
        ssl_kwargs["ssl_keyfile"] = key

    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=settings.port,
        **ssl_kwargs,
    )

Step 4: Run tests to verify they pass

Run: cd /home/edgarcosta/magma/calculator && . .venv/bin/activate && python -m pytest tests/test_main.py -v Expected: All PASS

Note: The custom CORS middleware replaces FastAPI's built-in CORSMiddleware to handle the localhost-any-port requirement cleanly. If the CORS tests fail, the middleware may need adjustment for how TestClient handles preflight requests.

Step 5: Commit

git add app/main.py tests/test_main.py
git commit -m "feat: FastAPI app with /execute and /health endpoints, CORS, rate limiting"

Task 7: Deploy scripts and Docker config

Files:

Create: deploy/certbot-renew-hook.sh
Create: nsjail.cfg
Create: Dockerfile
Create: docker-compose.yml

Step 1: Create certbot renewal hook

#!/bin/bash
# deploy/certbot-renew-hook.sh
# Install to: /etc/letsencrypt/renewal-hooks/post/restart-calculator.sh
# This script is called by certbot after successful certificate renewal.
docker restart magma-calculator

Make executable: chmod +x deploy/certbot-renew-hook.sh

Step 2: Create nsjail config

Note: The exact mount paths depend on the Magma installation and the host's shared library layout. The uidmap/gidmap needs to match the calculator user's UID/GID inside the container. This config will need testing and adjustment during Docker integration testing (Task 8).

name: "magma-sandbox"
description: "Sandbox for Magma calculator execution"

mode: ONCE
time_limit: 0
rlimit_as_type: SOFT
rlimit_cpu_type: SOFT

uidmap {
    inside_id: "calculator"
    outside_id: "calculator"
}
gidmap {
    inside_id: "calculator"
    outside_id: "calculator"
}

clone_newnet: true
clone_newpid: true
clone_newns: true
clone_newuts: true

mount {
    src: "/opt/magma"
    dst: "/opt/magma"
    is_bind: true
    rw: false
}
mount {
    dst: "/tmp"
    fstype: "tmpfs"
    rw: true
    options: "size=67108864"
}
mount {
    src: "/usr/lib"
    dst: "/usr/lib"
    is_bind: true
    rw: false
}
mount {
    src: "/lib"
    dst: "/lib"
    is_bind: true
    rw: false
}
mount {
    src: "/lib64"
    dst: "/lib64"
    is_bind: true
    rw: false
    mandatory: false
}
mount {
    src: "/etc/ld.so.cache"
    dst: "/etc/ld.so.cache"
    is_bind: true
    rw: false
    mandatory: false
}

envar: "PATH=/opt/magma/bin:/usr/bin:/bin"
envar: "HOME=/tmp"

keep_env: false

Step 3: Create Dockerfile

# Stage 1: Python dependencies
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user -r requirements.txt

# Stage 2: nsjail
FROM ubuntu:24.04 AS nsjail-builder
RUN apt-get update && apt-get install -y \
    git build-essential pkg-config \
    libprotobuf-dev protobuf-compiler libnl-3-dev libnl-route-3-dev && \
    rm -rf /var/lib/apt/lists/*
RUN git clone https://github.com/google/nsjail.git /nsjail && \
    cd /nsjail && make

# Stage 3: Runtime
FROM python:3.11-slim
WORKDIR /app

# Install nsjail runtime dependencies (verify package names for target distro)
RUN apt-get update && apt-get install -y --no-install-recommends \
    libnl-3-200 libnl-route-3-200 libprotobuf-lite32t64 && \
    rm -rf /var/lib/apt/lists/*

# Create non-root user for Magma (nsjail drops privileges to this user)
RUN useradd -m calculator

COPY --from=nsjail-builder /nsjail/nsjail /usr/local/bin/nsjail
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

COPY app/ ./app/
COPY nsjail.cfg .

# Runs as root (required for nsjail namespace creation)
# nsjail drops privileges to 'calculator' for Magma execution
EXPOSE 8080

CMD ["python", "-m", "app.main"]

Note: The libprotobuf-lite32t64 package name is for Ubuntu 24.04. If the build fails, check correct name with: docker run --rm ubuntu:24.04 apt-cache search libprotobuf-lite

Step 4: Create docker-compose.yml

services:
  calculator:
    build: .
    container_name: magma-calculator
    cap_add:
      - SYS_ADMIN
    tmpfs:
      - /tmp:size=128m
    volumes:
      - /opt/magma:/opt/magma:ro
      - /etc/letsencrypt:/certs:ro
    env_file:
      - calculator.env
    ports:
      - "443:8080"
    restart: unless-stopped

Step 5: Commit

git add deploy/certbot-renew-hook.sh nsjail.cfg Dockerfile docker-compose.yml
git commit -m "feat: Docker setup with nsjail config, Dockerfile, docker-compose, and certbot hook"

Task 8: Integration testing with Docker

This task verifies the full stack works. It requires a machine with Docker and a Magma installation.

Step 1: Build the Docker image

Run: cd /home/edgarcosta/magma/calculator && docker build -t magma-calculator .

If the build fails on libprotobuf-lite32t64, check the correct package name: Run: docker run --rm ubuntu:24.04 apt-cache search libprotobuf-lite Update the Dockerfile with the correct package name.

Step 2: Test without TLS first

Run the container without TLS to verify basic functionality. Set TLS_CERT_FILE and TLS_KEY_FILE to empty/nonexistent to skip TLS:

docker run --rm \
  --name magma-calculator-test \
  --cap-add SYS_ADMIN \
  --tmpfs /tmp:size=128m \
  -v /opt/magma:/opt/magma:ro \
  -e TLS_CERT_FILE=/nonexistent \
  -e TLS_KEY_FILE=/nonexistent \
  -p 8080:8080 \
  magma-calculator

In another terminal:

curl -X POST http://localhost:8080/execute \
  -H "Content-Type: application/json" \
  -d '{"code": "print 1+1;"}'

Expected: JSON response with "success": true, "stdout": "2\n", version, seed, time, memory.

Step 3: Test health endpoint

curl http://localhost:8080/health

Expected: {"status":"ok"}

Step 4: Test rate limiting

for i in $(seq 1 35); do
  curl -s -o /dev/null -w "%{http_code}\n" -X POST http://localhost:8080/execute \
    -H "Content-Type: application/json" \
    -d '{"code": "print 1;"}'
done

Expected: first 30 return 200, then 429.

Step 5: Test nsjail isolation

curl -X POST http://localhost:8080/execute \
  -H "Content-Type: application/json" \
  -d '{"code": "System(\"ls\");"}'

Expected: 200 with warning about error (Magma -w blocks System).

Step 6: Adjust nsjail.cfg as needed

The nsjail config will likely need adjustments for mount paths and library dependencies. Common issues:

Magma can't find shared libraries: add more library directories to mounts
Magma can't find its startup files: ensure mount includes everything needed
UID/GID mismatch: verify calculator user's IDs match between nsjail config and container

Step 7: Commit any fixes

git add -A
git commit -m "fix: adjust nsjail config and Dockerfile for integration testing"

edgarcosta/2026-02-04-magma-calculator-impl.md

Select an option

No results found

Select an option

No results found

Magma Calculator Service Implementation Plan

Task 1: Project scaffolding and config

Task 2: Magma output parser

Task 3: Stderr parser

Task 4: Rate limiter

Task 5: Executor (subprocess management)

Task 6: FastAPI application and endpoints

Task 7: Deploy scripts and Docker config

Task 8: Integration testing with Docker