diff --git a/test/test_composer_path_containment.py b/test/test_composer_path_containment.py new file mode 100644 index 00000000..cc5f067b --- /dev/null +++ b/test/test_composer_path_containment.py @@ -0,0 +1,91 @@ +"""A composer plugin id must never resolve outside the plugins directory. + +CodeQL reported sixteen high-severity py/path-injection alerts against +web_interface/blueprints/composer.py: a request-supplied plugin_id reaching +Path(plugins_dir) / plugin_id, which is then created, written to, deleted +(shutil.rmtree) and read back. + +The id was already validated by an anchored regex, so every traversal payload +was in fact rejected. What was missing was the guarantee living *with* the path +building rather than in a regex several hundred lines away -- loosen that regex +later and the traversal opens silently, with nothing at the filesystem boundary +to catch it. _plugin_dir() closes that, and is the form static analysis can see. +""" +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from web_interface.blueprints import composer as C # noqa: E402 + +#: Anything that has ever been used to climb out of a directory. +TRAVERSAL = [ + "../../etc/passwd", "..", ".", "a/../../etc", "good/../../..", + "/etc/passwd", "//etc/passwd", "a\\..\\..", "a%2f..%2f..", + "....//....//etc", "a/./../../etc", "~", "~root", + "plugin/../../../../../../etc/shadow", +] + +#: Rejected for shape, not traversal -- but rejected all the same. +MALFORMED = ["", "A-upper", "1-leading-digit", "-leading-dash", "has_underscore", + "has space", "has.dot", "a" * 64, "plugin\n", "plugin\n../../etc", + "\n", "plug\x00in"] + + +@pytest.fixture +def plugins_dir(tmp_path, monkeypatch): + base = tmp_path / "plugin-repos" + base.mkdir() + monkeypatch.setattr(C.composer_bp, "plugins_dir", str(base), raising=False) + return base + + +@pytest.mark.parametrize("payload", TRAVERSAL) +def test_traversal_payloads_are_refused(plugins_dir, payload): + with pytest.raises(C.ComposerInputError): + C._plugin_dir(payload) + + +@pytest.mark.parametrize("payload", MALFORMED) +def test_malformed_ids_are_refused(plugins_dir, payload): + with pytest.raises(C.ComposerInputError): + C._plugin_dir(payload) + + +@pytest.mark.parametrize("payload", ["a", "my-plugin", "x9", "a" * 63]) +def test_valid_ids_resolve_inside_the_base(plugins_dir, payload): + resolved = C._plugin_dir(payload) + assert resolved.parent == plugins_dir.resolve(), ( + f"{payload!r} resolved to {resolved}, outside {plugins_dir}") + + +def test_no_payload_can_escape_even_if_the_regex_is_loosened(plugins_dir, monkeypatch): + """The containment check must stand on its own. + + This is the whole point of resolving at the filesystem boundary: if the id + pattern is ever relaxed, traversal must still be impossible. Replace the + regex with one that permits slashes and dots, then re-run the payloads. + """ + import re + monkeypatch.setattr(C, "_PLUGIN_ID_RE", re.compile(r"\A[\w./\\~-]+\Z")) + escaped = [] + for payload in TRAVERSAL: + try: + resolved = C._plugin_dir(payload) + except C.ComposerInputError: + continue + base = plugins_dir.resolve() + if resolved != base and base not in resolved.parents: + escaped.append((payload, str(resolved))) + assert not escaped, f"these escaped the base with a loosened regex: {escaped}" + + +def test_a_trailing_newline_is_not_a_valid_id(): + r"""Python's `$` also matches before a trailing newline, so the original + `^...$` accepted "myplugin\n" and would have created a directory whose + name ends in one. \Z does not.""" + assert C._PLUGIN_ID_RE.match("myplugin") is not None + assert C._PLUGIN_ID_RE.match("myplugin\n") is None diff --git a/web_interface/blueprints/composer.py b/web_interface/blueprints/composer.py index 9b3461f6..3eb0002c 100644 --- a/web_interface/blueprints/composer.py +++ b/web_interface/blueprints/composer.py @@ -48,7 +48,11 @@ _FONT_SIZE_MAP = { 'five_by_seven': 7, } -_PLUGIN_ID_RE = re.compile(r'^[a-z][a-z0-9-]{0,62}$') +_PLUGIN_ID_RE = re.compile(r'\A[a-z][a-z0-9-]{0,62}\Z') +#: \Z, not $. Python's $ also matches just before a trailing newline, +#: so '$' would accept "myplugin\\n" and create a directory whose name +#: ends in one. Not traversal, but not a name anything should have to +#: handle either. _PYTHON_IDENT_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$') # ── Jinja2 environment (separate from Flask's; autoescape=False for code gen) ── @@ -431,18 +435,18 @@ def _generate_plugin_files(data: dict) -> dict: plugin_id = metadata.get('id', '').strip() if not _PLUGIN_ID_RE.match(plugin_id): - raise ValueError( + raise ComposerInputError( 'Plugin ID must start with a lowercase letter and contain only ' 'lowercase letters, numbers, and hyphens (max 63 chars).' ) plugin_name = metadata.get('name', '').strip() if not plugin_name: - raise ValueError('Plugin name is required.') + raise ComposerInputError('Plugin name is required.') author = metadata.get('author', '').strip() if not author: - raise ValueError('Author is required.') + raise ComposerInputError('Author is required.') version = metadata.get('version', '1.0.0').strip() @@ -450,7 +454,7 @@ def _generate_plugin_files(data: dict) -> dict: for cv in config_vars: key = cv.get('key', '') if not _PYTHON_IDENT_RE.match(key): - raise ValueError(f'Config variable key "{key}" is not a valid Python identifier.') + raise ComposerInputError(f'Config variable key "{key}" is not a valid Python identifier.') class_name = _to_class_name(plugin_name) # Only consider visible elements for code generation flags @@ -481,7 +485,7 @@ def _generate_plugin_files(data: dict) -> dict: try: tmpl = env.get_template('manager.py.j2') except jinja2.TemplateNotFound: - raise ValueError('Code generation template not found. This is a server configuration issue.') + raise ComposerInputError('Code generation template not found. This is a server configuration issue.') manager_py = tmpl.render( plugin_name=plugin_name, @@ -501,7 +505,7 @@ def _generate_plugin_files(data: dict) -> dict: try: ast.parse(manager_py) except SyntaxError as exc: - raise ValueError(f'Generated code has a syntax error: {exc}') from exc + raise ComposerInputError(f'Generated code has a syntax error: {exc}') from exc # Build manifest manifest = { @@ -533,7 +537,7 @@ def _generate_plugin_files(data: dict) -> dict: errors = list(validator.iter_errors(manifest)) if errors: msgs = '; '.join(e.message for e in errors[:3]) - raise ValueError(f'Manifest validation failed: {msgs}') + raise ComposerInputError(f'Manifest validation failed: {msgs}') # Build config_schema type_map = { @@ -585,6 +589,37 @@ def _generate_plugin_files(data: dict) -> dict: } +class ComposerInputError(ValueError): + """A validation failure whose message is safe to show the caller. + + _generate_plugin_files raises this for input the user can fix. Anything + else reaching the handlers is unexpected, and its text may name internal + paths or library internals, so it is logged and answered generically. + """ + + +def _plugin_dir(plugin_id: str) -> Path: + """Resolve a plugin directory, refusing anything outside plugins_dir. + + _PLUGIN_ID_RE already rejects '/', '.' and '..', so this cannot currently + fail -- every traversal payload is blocked before it gets here. It exists + anyway for two reasons: the guarantee then lives with the path building + rather than in a regex several hundred lines away, so loosening that regex + later cannot silently open a traversal; and it is the form static analysis + recognises, which is why CodeQL reported sixteen path-injection alerts + against code that was already safe. + + Raises ComposerInputError if the id is malformed or escapes the base. + """ + if not _PLUGIN_ID_RE.match(plugin_id or ''): + raise ComposerInputError('Invalid plugin ID') + base = Path(composer_bp.plugins_dir).resolve() + candidate = (base / plugin_id).resolve() + if candidate != base and base not in candidate.parents: + raise ComposerInputError('Invalid plugin ID') + return candidate + + def _save_composer_state(target_dir: Path, payload: dict) -> None: """Persist the raw composer payload alongside the generated plugin files.""" (target_dir / '_composer_state.json').write_text( @@ -618,8 +653,13 @@ def generate_zip(): return jsonify({'status': 'error', 'message': 'No JSON body'}), 400 try: files = _generate_plugin_files(data) - except ValueError as exc: + except ComposerInputError as exc: return jsonify({'status': 'error', 'message': str(exc)}), 422 + except ValueError as exc: + # Not one of ours: the text may name internal paths or library + # internals, so log it and answer generically. + logger.exception('Unexpected error generating plugin files: %s', exc) + return jsonify({'status': 'error', 'message': 'Could not generate plugin files'}), 422 plugin_id = data.get('metadata', {}).get('id', 'plugin') files['_composer_state.json'] = json.dumps(data, indent=2, ensure_ascii=False) @@ -643,8 +683,13 @@ def install_locally(): try: files = _generate_plugin_files(data) - except ValueError as exc: + except ComposerInputError as exc: return jsonify({'status': 'error', 'message': str(exc)}), 422 + except ValueError as exc: + # Not one of ours: the text may name internal paths or library + # internals, so log it and answer generically. + logger.exception('Unexpected error generating plugin files: %s', exc) + return jsonify({'status': 'error', 'message': 'Could not generate plugin files'}), 422 plugin_id = data.get('metadata', {}).get('id', '') # _generate_plugin_files() above already validates metadata.id via this @@ -652,9 +697,10 @@ def install_locally(): # different function -- re-check here, at the point the path is actually # built, so this route stays safe on its own if that call is ever # reordered or changed. - if not _PLUGIN_ID_RE.match(plugin_id): - return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400 - target = Path(composer_bp.plugins_dir) / plugin_id + try: + target = _plugin_dir(plugin_id) + except ComposerInputError as exc: + return jsonify({'status': 'error', 'message': str(exc)}), 400 force = bool(data.get('_force', False)) if target.exists() and not force: @@ -708,7 +754,10 @@ def validate_id(plugin_id): if not _PLUGIN_ID_RE.match(plugin_id): return jsonify({'valid': False, 'available': False, 'reason': 'Invalid format'}) if composer_bp.plugins_dir: - taken = (Path(composer_bp.plugins_dir) / plugin_id).exists() + try: + taken = _plugin_dir(plugin_id).exists() + except ComposerInputError: + return jsonify({'valid': False, 'available': False, 'reason': 'Invalid format'}) if taken: return jsonify({'valid': True, 'available': False, 'reason': 'Already installed'}) return jsonify({'valid': True, 'available': True}) @@ -751,8 +800,13 @@ def preview_code(): return jsonify({'status': 'error', 'message': 'No JSON body'}), 400 try: files = _generate_plugin_files(data) - except ValueError as exc: + except ComposerInputError as exc: return jsonify({'status': 'error', 'message': str(exc)}), 422 + except ValueError as exc: + # Not one of ours: the text may name internal paths or library + # internals, so log it and answer generically. + logger.exception('Unexpected error generating plugin files: %s', exc) + return jsonify({'status': 'error', 'message': 'Could not generate plugin files'}), 422 return jsonify({ 'status': 'ok', 'files': { @@ -772,10 +826,10 @@ def load_plugin(plugin_id): """ if not composer_bp.plugins_dir: return jsonify({'status': 'error', 'message': 'Plugin directory not configured'}), 503 - if not _PLUGIN_ID_RE.match(plugin_id): - return jsonify({'status': 'error', 'message': 'Invalid plugin ID'}), 400 - - plugin_dir = Path(composer_bp.plugins_dir) / plugin_id + try: + plugin_dir = _plugin_dir(plugin_id) + except ComposerInputError as exc: + return jsonify({'status': 'error', 'message': str(exc)}), 400 if not plugin_dir.exists(): return jsonify({'status': 'error', 'message': 'Plugin not found'}), 404