Resolve analytic functions by registry name or allowed module prefix (ISSUE-169)

PYEGERIA_ISSUES.md

@@ -172,6 +172,7 @@ re-test candidates moved into *Awaiting re-test* below the open sections.
   create-from-template) and 7 (`viewElementGUID`) are still open.
 
 **Open pyegeria items**
+- ISSUE-169: `_resolve_analytic_function` imported any dotted path a report spec named; now registered names plus an allowlist of module prefixes
 - ISSUE-168: pyegeria can't run on Python 3.14: `nest_asyncio`, applied at import time, makes `asyncio.current_task()` return None
 - ISSUE-155: pyegeria gaps against Egeria `main` since 6.2 (omvs_audit + source diff, V6.2..05a5e982)
 - ISSUE-150: Dr.Egeria commands for an Analytic Function's Egeria declaration: a GovernanceActionType that Resource Explorer runs as an engine host, plus a synchronous path that bypasses Egeria
@@ -193,6 +194,24 @@ into the entry now, so it isn't rediscovered from scratch later.
 
 ---
 
+### ISSUE-169: `_resolve_analytic_function` imported any dotted path a report spec named; now registered names plus an allowlist of module prefixes
+
+**Layer:** pyegeria · **Status:** fix in PR (branch `fix/analytic-resolver-allowlist`) · **Found:** 2026-10-30, Resource Explorer's portal-extensibility evaluation (gap G6, trellis `docs/design-notes/EVAL-PORTAL-EXTENSIBILITY-VS-RE-2026-10-30.md`).
+
+**What:** `format_set_executor._resolve_analytic_function` ran `importlib.import_module` on whatever module a report spec's `analytic_function` (or an action's `fetch`/`analytic` step, or an `analytic_binders` factory) named. A report spec is data, which can come from a user's report-spec directory or a config tier, so it could make the executor import and call any module on the path.
+
+**Fix (owner's option C, kept quiet: no deprecation warning yet):**
+1. A registered name (`analytic_registry`) resolves to that entry's `function`.
+2. A dotted path some registered entry declares is trusted as before.
+3. Any other dotted path must sit under an allowed module prefix: `PYEGERIA_ANALYTIC_MODULE_PREFIXES` (comma-separated), default `pyegeria`, `trellis_analytics`. A prefix is a package boundary, so `pyegeria` does not admit `pyegeria_extras`. Such a path logs at DEBUG only.
+4. Anything else raises `AnalyticFunctionRefused` (a `ValueError`) before any import.
+
+No downstream spec in egeria-workspaces or trellis names a module outside the default list (checked 2026-10-30). Tests: `tests/micro-tests/test_issue_169_analytic_resolver.py`.
+
+**Later, not now:** once entry-point registration exists (ISSUE-172), consider a deprecation warning for unregistered dotted paths, then registry-only resolution.
+
+---
+
 ### ISSUE-168: pyegeria can't run on Python 3.14: `nest_asyncio`, applied at import time, makes `asyncio.current_task()` return None
 
 **Layer:** pyegeria · **Status:** fix drafted on branch `fix/issue-168-py314-sync-runner` (not merged) · **Found:** 2026-10-29, when the owner asked why pyegeria caps Python at `<3.14`.

pyegeria/view/format_set_executor.py

@@ -16,6 +16,7 @@
 import asyncio
 import inspect
 import json
+import os
 import sys
 from typing import Any, Dict, Optional, Union
 
@@ -85,15 +86,31 @@ def _resolve_client_and_method(func_decl: str):
     return (client_class, method_name)
 
 
-def _resolve_analytic_function(func_decl: str):
-    """Given a dotted import path like 'pyegeria.view.overview_metrics.growth_series',
-    import and return the callable. Unlike `function` (find_method), `analytic_function`
-    (extra_find) targets a plain module-level function, not a client class method --
-    overview_metrics.py's whole design boundary is that its functions take an
-    already-constructed, already-authenticated client as their first argument, so
-    there's no client class to resolve here, just the function itself."""
-    if not isinstance(func_decl, str) or "." not in func_decl:
-        raise ValueError(f"Invalid analytic_function declaration: {func_decl!r}")
+# Module prefixes a report spec may name directly as a dotted path (ISSUE-169).
+# Override with PYEGERIA_ANALYTIC_MODULE_PREFIXES (comma-separated). A prefix
+# matches the module itself or anything under it, so "pyegeria" does not
+# admit "pyegeria_extras".
+DEFAULT_ANALYTIC_MODULE_PREFIXES = ("pyegeria", "trellis_analytics")
+
+
+class AnalyticFunctionRefused(ValueError):
+    """A report spec named a callable that is neither registered nor under an allowed module prefix."""
+    pass
+
+
+def analytic_module_prefixes() -> tuple:
+    """The allowed module prefixes: PYEGERIA_ANALYTIC_MODULE_PREFIXES if set, else the defaults."""
+    raw = os.getenv("PYEGERIA_ANALYTIC_MODULE_PREFIXES", "").strip()
+    if not raw:
+        return DEFAULT_ANALYTIC_MODULE_PREFIXES
+    return tuple(p.strip().rstrip(".") for p in raw.split(",") if p.strip())
+
+
+def _module_allowed(module_path: str) -> bool:
+    return any(module_path == p or module_path.startswith(p + ".") for p in analytic_module_prefixes())
+
+
+def _import_callable(func_decl: str):
     module_path, func_name = func_decl.rsplit(".", 1)
     module = importlib.import_module(module_path)
     func = getattr(module, func_name, None)
@@ -102,6 +119,46 @@ def _resolve_analytic_function(func_decl: str):
     return func
 
 
+def _resolve_analytic_function(func_decl: str):
+    """Return the callable a report spec's analytic_function (or an action's
+    fetch/analytic step, or a binder factory) names.
+
+    Unlike `function` (find_method), `analytic_function` (extra_find) targets a
+    plain module-level function, not a client class method -- overview_metrics.py's
+    whole design boundary is that its functions take an already-constructed,
+    already-authenticated client as their first argument, so there's no client
+    class to resolve here, just the function itself.
+
+    A report spec is data, so it may not import arbitrary code (ISSUE-169):
+      1. A registered name (analytic_registry) resolves to that entry's function.
+      2. A dotted path some registered entry declares is trusted as-is.
+      3. Any other dotted path must sit under an allowed module prefix
+         (`analytic_module_prefixes()`); anything else raises
+         AnalyticFunctionRefused before any import happens.
+    """
+    if not isinstance(func_decl, str) or not func_decl.strip():
+        raise ValueError(f"Invalid analytic_function declaration: {func_decl!r}")
+    from pyegeria.view.analytic_registry import get_analytic_registry
+
+    registry = get_analytic_registry()
+    spec = registry.get(func_decl)
+    if spec is not None:
+        return _import_callable(spec.function)
+    if "." not in func_decl:
+        raise AnalyticFunctionRefused(
+            f"Analytic function '{func_decl}' is not a registered name, and is not a dotted path.")
+    if any(s.function == func_decl for s in registry.values()):
+        return _import_callable(func_decl)
+    module_path = func_decl.rsplit(".", 1)[0]
+    if not _module_allowed(module_path):
+        raise AnalyticFunctionRefused(
+            f"Analytic function '{func_decl}' is not registered and its module is not under an allowed "
+            f"prefix {list(analytic_module_prefixes())}. Register it in the analytic registry, or add its "
+            f"package to PYEGERIA_ANALYTIC_MODULE_PREFIXES.")
+    logger.debug(f"Analytic function '{func_decl}' resolved by dotted path (not registered).")
+    return _import_callable(func_decl)
+
+
 # Parameter names an analytic function uses for its leading client argument(s)
 # -- pyegeria.view.overview_metrics's functions take one (`mgr` or `ce`) or,
 # for semantic_grounding, two (`mgr`, `ce`) positional clients before their

tests/micro-tests/test_issue_138_140_149_analytic_envelope.py

@@ -74,6 +74,7 @@ def fake_module(monkeypatch):
     for f in (Reader, read_with, measured, not_established, nothing_found, not_applicable, plain_counts, model_result):
         setattr(mod, f.__name__, f)
     monkeypatch.setitem(sys.modules, MOD, mod)
+    monkeypatch.setenv("PYEGERIA_ANALYTIC_MODULE_PREFIXES", f"pyegeria,{MOD}")  # ISSUE-169
 
     def no_egeria(*a, **k):
         raise AssertionError("an analytic routine with no client must not build an Egeria client")

tests/micro-tests/test_issue_169_analytic_resolver.py

@@ -0,0 +1,99 @@
+"""ISSUE-169: a report spec's analytic_function may only name a registered function
+or a dotted path under an allowed module prefix; anything else is refused before import."""
+import sys
+
+import pytest
+
+import pyegeria.view.format_set_executor as fse
+from pyegeria.core.utils import body_slimmer
+from pyegeria.view import overview_metrics
+from pyegeria.view.analytic_registry import (
+    AnalyticFunctionSpec,
+    clear_runtime_analytic_functions,
+    register_analytic_functions,
+)
+
+
+@pytest.fixture(autouse=True)
+def _clean(monkeypatch):
+    monkeypatch.delenv("PYEGERIA_ANALYTIC_MODULE_PREFIXES", raising=False)
+    clear_runtime_analytic_functions()
+    yield
+    clear_runtime_analytic_functions()
+
+
+@pytest.fixture
+def import_calls(monkeypatch):
+    calls = []
+    real = fse.importlib.import_module
+
+    def _recording(name, *args, **kwargs):
+        calls.append(name)
+        return real(name, *args, **kwargs)
+
+    monkeypatch.setattr(fse.importlib, "import_module", _recording)
+    return calls
+
+
+def test_registered_name_resolves_to_its_function():
+    register_analytic_functions({"test-169-growth": AnalyticFunctionSpec(
+        name="test-169-growth", function="pyegeria.view.overview_metrics.growth_series")})
+    assert fse._resolve_analytic_function("test-169-growth") is overview_metrics.growth_series
+
+
+def test_registered_function_outside_allowlist_is_trusted(monkeypatch):
+    monkeypatch.setenv("PYEGERIA_ANALYTIC_MODULE_PREFIXES", "nothing_matches")
+    register_analytic_functions({"test-169-json": AnalyticFunctionSpec(name="test-169-json", function="json.dumps")})
+    import json
+    assert fse._resolve_analytic_function("test-169-json") is json.dumps
+    assert fse._resolve_analytic_function("json.dumps") is json.dumps
+
+
+def test_allowed_dotted_path_resolves_and_logs_at_debug_only():
+    from loguru import logger
+    records = []
+    sink = logger.add(lambda m: records.append(m.record), level="DEBUG")
+    try:
+        func = fse._resolve_analytic_function("pyegeria.core.utils.body_slimmer")
+    finally:
+        logger.remove(sink)
+    assert func is body_slimmer
+    mine = [r for r in records if "body_slimmer" in r["message"]]
+    assert mine and all(r["level"].name == "DEBUG" for r in mine)
+
+
+def test_prefix_is_a_package_boundary_not_a_string_prefix(import_calls):
+    with pytest.raises(fse.AnalyticFunctionRefused):
+        fse._resolve_analytic_function("pyegeria_extras.mod.func")
+    assert import_calls == []
+
+
+def test_refused_path_raises_and_never_imports(import_calls):
+    sys.modules.pop("test_169_never_imported", None)
+    with pytest.raises(fse.AnalyticFunctionRefused, match="not under an allowed prefix"):
+        fse._resolve_analytic_function("test_169_never_imported.evil")
+    with pytest.raises(fse.AnalyticFunctionRefused):
+        fse._resolve_analytic_function("os.system")
+    assert import_calls == []
+    assert "test_169_never_imported" not in sys.modules
+
+
+def test_unregistered_bare_name_is_refused(import_calls):
+    with pytest.raises(fse.AnalyticFunctionRefused, match="not a registered name"):
+        fse._resolve_analytic_function("no-such-function")
+    assert import_calls == []
+
+
+def test_env_var_overrides_the_allowlist(monkeypatch):
+    monkeypatch.setenv("PYEGERIA_ANALYTIC_MODULE_PREFIXES", "json, trellis_analytics.")
+    import json
+    assert fse._resolve_analytic_function("json.dumps") is json.dumps
+    with pytest.raises(fse.AnalyticFunctionRefused):
+        fse._resolve_analytic_function("pyegeria.core.utils.body_slimmer")
+
+
+def test_binder_factories_go_through_the_same_check():
+    def analytic(ods=None):
+        return ods
+    with pytest.raises(fse.AnalyticFunctionRefused):
+        fse._apply_analytic_binders(analytic, {"ods": "os:getcwd"}, {})