diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2236091..09e76f6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -38,9 +38,8 @@ jobs: # pytest run cross-contaminates sys.modules. So run each in its own # process with the inifile addopts (pytest-django flags) cleared. # Excluded: test_integration (needs a live backend), - # test_audit_trail (pulls full app deps incl. psycopg), - # test_dynamic_survey_standalone (order-dependent Django stub). - for f in $(ls tests_standalone/test_*.py | grep -vE 'test_integration|test_audit_trail|test_dynamic_survey_standalone'); do + # test_audit_trail (pulls full app deps incl. psycopg). + for f in $(ls tests_standalone/test_*.py | grep -vE 'test_integration|test_audit_trail'); do echo "::group::$f" python -m pytest "$f" -o addopts="" -q echo "::endgroup::" diff --git a/docs/13-dynamic-surveys.md b/docs/13-dynamic-surveys.md index 596ff67..220a030 100644 --- a/docs/13-dynamic-surveys.md +++ b/docs/13-dynamic-surveys.md @@ -15,7 +15,7 @@ surveys resolve choices from three configurable sources: | ------------------- | -------------------------------------------------- | ----------------------------------- | | **Database Query** | Query Forail models (hosts, groups, projects, etc.) | Select a host from inventory | | **External API** | Fetch choices from an HTTP endpoint | Options from CMDB, ServiceNow, etc. | -| **Jinja2 Template** | Evaluate a Jinja2 expression | Custom logic using inventory data | +| ~~Jinja2 Template~~ | **Withdrawn** — see below | — | Results are cached with a configurable TTL to avoid slow launches. @@ -60,7 +60,7 @@ A survey question with dynamic choices includes a `dynamic_choices` field: | Field | Type | Required | Description | | ------------- | ------- | ---------------- | -------------------------------------------- | | `enabled` | boolean | Yes | Enable/disable dynamic choices | -| `source_type` | string | Yes (if enabled) | One of: `db_query`, `api_endpoint`, `jinja2` | +| `source_type` | string | Yes (if enabled) | One of: `db_query`, `api_endpoint` (`jinja2` is withdrawn) | | `cache_ttl` | integer | No (default: 60) | Cache duration in seconds (0 = no cache) | --- @@ -111,7 +111,38 @@ the system automatically filters by the job template's inventory. ## Source: External API -Fetch choices from an HTTP endpoint. Supports JSON responses. +Fetch choices from an HTTPS endpoint. Supports JSON responses. + +> **The destination must be named by an administrator.** The server makes this +> request from inside the cluster, and the URL comes from the survey — so +> without a destination policy, whoever can edit a job template can point the +> server at cloud metadata, at a service bound to loopback, or at a neighbouring +> pod, and read the reply through the choices endpoint. Hosts are therefore +> listed in `SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST` in the server settings file, +> matched exactly. The list is **empty by default, which disables this source +> type**. +> +> ```python +> # /etc/tower/conf.d/dynamic_surveys.py +> SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST = ['cmdb.internal.example.com'] +> ``` +> +> Rules that apply even to a listed host: +> +> - **HTTPS only.** +> - The name is re-checked **after DNS resolution**. Private addresses are +> allowed — an on-prem CMDB on `10.0.0.0/8` is the ordinary case — but +> loopback, link-local (`169.254.0.0/16`, where cloud metadata lives), +> multicast and reserved addresses are refused. A listed name whose DNS answer +> points inward does not get through. +> - **Redirects are never followed.** The first hop is the one that was checked; +> every hop after it would be the peer's choice. +> - Methods are limited to `GET` and `POST`, `timeout` is capped at 30 seconds, +> and the response is read up to 1 MiB. +> +> A URL that is not permitted is rejected when the survey is **saved**, so the +> editor is told rather than left with a question that silently resolves to +> nothing. ```json { @@ -131,12 +162,12 @@ Fetch choices from an HTTP endpoint. Supports JSON responses. | Field | Type | Default | Description | | ------------- | ------- | ------- | ------------------------------------------ | -| `url` | string | — | HTTP endpoint URL (required) | +| `url` | string | — | HTTPS endpoint URL (required, host must be allowlisted) | | `method` | string | `GET` | HTTP method (`GET` or `POST`) | -| `headers` | object | `{}` | Custom HTTP headers | +| `headers` | object | `{}` | Custom HTTP headers. **Stored in the survey spec in plaintext** — anyone who can read the job template can read them | | `json_path` | string | `""` | Dot-notation path to the array in response | | `value_field` | string | `""` | Field to extract from objects in the array | -| `timeout` | integer | `10` | Request timeout in seconds | +| `timeout` | integer | `10` | Request timeout in seconds (capped at 30) | | `body` | object | `{}` | Request body for POST method | ### Response Formats @@ -165,15 +196,35 @@ With `json_path: "data.items"` and `value_field: "hostname"`, this returns --- -## Source: Jinja2 Template +## Source: Jinja2 Template — withdrawn -Evaluate a Jinja2 expression that outputs a JSON array. +**This source type is disabled and surveys can no longer be saved with it.** + +The template was rendered on the server, in the web process, whenever a user +with `start` permission opened the launch prompt. Jinja2 seeds every environment +with objects whose `__init__.__globals__` reaches Python's module table, so +whoever could edit a job template's survey could run arbitrary code as the web +process — `{{ cycler.__init__.__globals__.os.environ.get('PATH') }}` returned the +server's `PATH`. Earlier versions of this page claimed the templates ran in a +restricted sandbox. They did not. + +Surveys already stored with `source_type: jinja2` resolve to **no choices** and +log a warning; they are not executed. Move them to `db_query` (for anything +drawn from inventory) or `api_endpoint` (for anything computed elsewhere). + +An operator who accepts the risk can set +`SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED = True` in the server settings file — not +via `/api/v2/settings/`, deliberately, so the API used to store a template cannot +also enable its execution. Templates then render in a Jinja2 sandbox with globals +removed and a reduced filter set. Treat that as hardening, not as a boundary: +sandbox escapes are found periodically, and a template rendering in the web +process is worth an escape to whoever plants it. ```json { "enabled": true, "source_type": "jinja2", - "template": "{{ groups | tojson }}", + "template": "{{ groups | sort | tojson }}", "cache_ttl": 60 } ``` @@ -196,8 +247,8 @@ The template **must output a valid JSON array**. {# Filter hosts by prefix #} {{ hosts | select("match", "^web") | list | tojson }} -{# Static list generated from range #} -{{ range(1, 11) | list | tojson }} +{# `range` and the other Jinja globals are removed; build from context instead #} +{{ groups | sort | tojson }} ``` --- @@ -246,12 +297,17 @@ Requires `start` permission on the job template (same as launching). 1. `dynamic_choices` is only valid on `multiplechoice` and `multiselect` types 2. When `dynamic_choices.enabled` is `true`, static `choices` field is not required -3. During job launch, answers to dynamic choice questions skip static choice validation -4. The `source_type` must be one of: `db_query`, `api_endpoint`, `jinja2` +3. During job launch, an answer to a dynamic-choices question is checked against + the **resolved** list, not against the static `choices` field. If the source + cannot be resolved the list is empty and every answer is rejected — an answer + that cannot be validated is not accepted +4. The `source_type` must be one of: `db_query`, `api_endpoint`. `jinja2` is + rejected unless `SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED` is `True` in the + server settings file 5. DB query `model` must be from the allowed list 6. DB query `field` must be from: `name`, `id`, `description` 7. API endpoint requires a non-empty `url` -8. Jinja2 requires a non-empty `template` +8. Jinja2, where an operator has re-enabled it, requires a non-empty `template` 9. `cache_ttl` must be a non-negative integer --- @@ -271,16 +327,24 @@ Requires `start` permission on the job template (same as launching). ## Caching - Resolved choices are cached in Django's cache backend (Redis) -- Cache key includes the variable name and full source configuration hash +- The cache key covers the question, the source configuration, **and the scope + the answer was resolved in** — job template, organization and inventory. It + used to be the variable name plus a config hash alone, so two templates with + the same question and different inventories shared one entry and the second + caller was served the first one's host names +- Nothing is cached when there is no template to scope by - Default TTL: 60 seconds - Set `cache_ttl: 0` to disable caching -- Cache is shared across all users and launch requests +- Within one scope, the cache is shared across users --- ## Limitations -- Maximum 500 choices returned per question (to prevent UI issues) -- Jinja2 templates run in a restricted sandbox (no file I/O) +- Maximum 500 choices returned per question (to prevent UI issues), applied to + every source type +- Jinja2 as a source type is withdrawn; where an operator has re-enabled it, + templates render in a Jinja2 sandbox with globals removed and a reduced filter + set — hardening, not a trust boundary - External API requests have a configurable timeout (default 10s) - DB query filters are limited to safe field lookups for security diff --git a/forail/main/models/mixins.py b/forail/main/models/mixins.py index 09cd56e..1533faa 100644 --- a/forail/main/models/mixins.py +++ b/forail/main/models/mixins.py @@ -279,30 +279,29 @@ def _survey_element_validation(self, survey_element, data, validate_required=Tru if type(data[survey_element['variable']]) != list: errors.append("'%s' value is expected to be a list." % survey_element['variable']) else: - # Skip static choice validation for dynamic choices questions - dc = survey_element.get('dynamic_choices', {}) - if dc and dc.get('enabled'): - pass # Dynamic choices are validated at resolve time - else: - choice_list = copy(survey_element['choices']) - if isinstance(choice_list, str): - choice_list = [choice for choice in choice_list.splitlines() if choice.strip() != ''] - for val in data[survey_element['variable']]: - if val not in choice_list: - errors.append("Value %s for '%s' expected to be one of %s." % (val, survey_element['variable'], choice_list)) + choice_list, is_dynamic = self._survey_choice_list(survey_element) + for val in data[survey_element['variable']]: + if val not in choice_list: + errors.append(self._choice_error(survey_element, val, choice_list, is_dynamic)) elif survey_element['type'] == 'multiplechoice': - dc = survey_element.get('dynamic_choices', {}) - if dc and dc.get('enabled'): - pass # Dynamic choices are validated at resolve time - else: - choice_list = copy(survey_element['choices']) - if isinstance(choice_list, str): - choice_list = [choice for choice in choice_list.splitlines() if choice.strip() != ''] - if survey_element['variable'] in data: - if data[survey_element['variable']] not in choice_list: - errors.append("Value %s for '%s' expected to be one of %s." % (data[survey_element['variable']], survey_element['variable'], choice_list)) + choice_list, is_dynamic = self._survey_choice_list(survey_element) + if survey_element['variable'] in data: + if data[survey_element['variable']] not in choice_list: + errors.append(self._choice_error(survey_element, data[survey_element['variable']], choice_list, is_dynamic)) return errors + def _survey_choice_list(self, survey_element): + """Permitted values for a choice question; see the service for the why.""" + from forail.main.services.dynamic_survey import survey_choice_list + + return survey_choice_list(survey_element, template=self) + + @staticmethod + def _choice_error(survey_element, value, choice_list, is_dynamic): + from forail.main.services.dynamic_survey import choice_error_message + + return choice_error_message(survey_element, value, choice_list, is_dynamic) + def _accept_or_ignore_variables(self, data, errors=None, _exclude_errors=(), extra_passwords=None): survey_is_enabled = self.survey_enabled and self.survey_spec extra_vars = data.copy() diff --git a/forail/main/services/dynamic_survey.py b/forail/main/services/dynamic_survey.py index fb740a4..a524b5f 100644 --- a/forail/main/services/dynamic_survey.py +++ b/forail/main/services/dynamic_survey.py @@ -1,15 +1,160 @@ import hashlib +import ipaddress import json import logging +import socket import time +from urllib.parse import urlsplit import requests from django.apps import apps +from django.conf import settings from django.core.cache import cache logger = logging.getLogger('forail.main.services.dynamic_survey') DYNAMIC_CHOICES_CACHE_PREFIX = 'dynamic_survey_choices_' + +# Jinja2 as a choices source is a code-execution surface, not a formatting +# convenience: the template text is supplied by whoever may edit a job +# template's survey spec, and it is rendered inside the web process when any +# user with `start` permission opens the launch prompt. Rendering it through a +# plain Environment hands that user Python -- `{{ cycler.__init__.__globals__ }}` +# is the standard route from a template to the module table. +# +# So the source type is refused unless an operator turns it on, and the switch +# is a settings *file* value on purpose -- deliberately not a database-backed +# setting exposed over /api/v2/settings/. The same API surface used to plant a +# payload must not also be able to enable its execution. +SURVEY_DYNAMIC_CHOICES_JINJA2_SETTING = 'SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED' + +# Filters left reachable when an operator does enable the source type. Jinja2 +# seeds an environment with far more than a choices list needs; what is not +# here is removed rather than trusted to the sandbox. +JINJA2_ALLOWED_FILTERS = frozenset( + { + 'batch', + 'default', + 'first', + 'int', + 'join', + 'last', + 'length', + 'list', + 'lower', + 'map', + 'reject', + 'replace', + 'reverse', + 'select', + 'sort', + 'string', + 'tojson', + 'trim', + 'unique', + 'upper', + } +) + +# Same ceiling the DB source applies, for the same reason: a choices list is a +# dropdown, not a data export. +MAX_CHOICES = 500 + +# The api_endpoint source makes the server issue a request to a URL taken from +# the survey. Without a destination policy that is an SSRF primitive: whoever +# edits a job template picks the address, and the server reaches it from inside +# the cluster -- cloud metadata endpoints, admin ports bound to loopback, +# neighbouring services -- while the JSON body comes back through the choices +# endpoint to any user with `start` permission. +# +# The destination must therefore be named by an operator, in a settings file, +# and it is host-exact: no wildcards, no suffix matching. An empty list (the +# default) disables the source type outright. +SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST_SETTING = 'SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST' + +# Even an allowlisted name is re-checked after resolution, which is what stops an +# allowlisted host from being pointed at the metadata service. Private ranges are +# deliberately permitted: an on-prem CMDB on 10.0.0.0/8 is the ordinary case for +# this feature, and the operator naming the host is the trust decision. What is +# refused is the set no legitimate choices API lives on. +API_FORBIDDEN_MESSAGE = { + 'loopback': 'a loopback address', + 'link_local': 'a link-local address (this is where cloud metadata lives)', + 'multicast': 'a multicast address', + 'reserved': 'a reserved address', + 'unspecified': 'the unspecified address', +} + +# A choices list that does not fit in a megabyte is not a choices list. Read +# bounded rather than trusting Content-Length, which the peer controls. +API_RESPONSE_MAX_BYTES = 1024 * 1024 + +# The survey supplies the timeout, so it needs a ceiling: a request that hangs +# holds a web worker for as long as it is allowed to. +API_MAX_TIMEOUT = 30 + + +def _api_allowlist(): + hosts = getattr(settings, SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST_SETTING, ()) or () + if isinstance(hosts, str): + hosts = (hosts,) + try: + return {h.strip().lower() for h in hosts if isinstance(h, str) and h.strip()} + except TypeError: + return set() + + +def api_destination_refusal(url): + """ + Why this URL may not be fetched, or None if it may. + + Returns a reason string rather than a bool so the caller can log which rule + refused; the reason is deliberately not returned to the API client, since it + would otherwise answer questions about the internal network. + """ + try: + parts = urlsplit(url) + except ValueError: + return 'the URL cannot be parsed' + + if parts.scheme != 'https': + return f"the scheme must be https, got {parts.scheme or 'none'}" + + host = parts.hostname + if not host: + return 'the URL has no host' + + allow = _api_allowlist() + if not allow: + return f'{SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST_SETTING} is empty, so no destination is permitted' + if host.lower() not in allow: + return f'the host is not listed in {SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST_SETTING}' + + try: + infos = socket.getaddrinfo(host, parts.port or 443, proto=socket.IPPROTO_TCP) + except OSError: + return 'the host does not resolve' + + for info in infos: + try: + addr = ipaddress.ip_address(info[4][0]) + except ValueError: + return 'the host resolves to an address that cannot be parsed' + for attribute, description in API_FORBIDDEN_MESSAGE.items(): + if getattr(addr, f'is_{attribute}', False): + return f'the host resolves to {description}' + return None + + +def jinja2_source_enabled(): + """ + Whether the jinja2 dynamic_choices source type may be used. + + The comparison is `is True` rather than a truth test on purpose: a stray + string, a `1`, or a test double standing in for the settings object must + not be enough to turn a code-execution path back on. + """ + return getattr(settings, SURVEY_DYNAMIC_CHOICES_JINJA2_SETTING, False) is True ALLOWED_DB_MODELS = { 'hosts': ('main', 'Host'), 'groups': ('main', 'Group'), @@ -24,9 +169,35 @@ ALLOWED_DB_FIELDS = {'name', 'id', 'description'} -def _cache_key(question_variable, source_config): - config_hash = hashlib.md5(json.dumps(source_config, sort_keys=True).encode()).hexdigest() - return f'{DYNAMIC_CHOICES_CACHE_PREFIX}{question_variable}_{config_hash}' +def _cache_key(question_variable, source_config, template): + """ + Cache key for one question's resolved choices. + + The key must carry the scope the answer was resolved *in*, not just the + question and its config. Two job templates can hold the same question with + the same `dynamic_choices` block and different inventories -- or belong to + different organizations -- and a key made of the variable name and a config + hash alone is identical for both. The first request then fills the cache + with one tenant's host names and the second is served them. + + Returns None when there is no template to scope by, which the caller treats + as "do not cache" rather than "cache globally". + """ + if template is None: + return None + + scope = ( + getattr(template, 'pk', None), + getattr(template, 'organization_id', None), + getattr(template, 'inventory_id', None), + ) + if scope[0] is None: + return None + + digest = hashlib.sha256( + json.dumps([question_variable, source_config, scope], sort_keys=True, default=str).encode() + ).hexdigest() + return f'{DYNAMIC_CHOICES_CACHE_PREFIX}{digest}' def resolve_dynamic_choices(question, template=None): @@ -44,10 +215,11 @@ def resolve_dynamic_choices(question, template=None): variable = question.get('variable', '') # Check cache - ck = _cache_key(variable, dc) - cached = cache.get(ck) - if cached is not None: - return cached + ck = _cache_key(variable, dc, template) + if ck is not None: + cached = cache.get(ck) + if cached is not None: + return cached choices = [] try: @@ -56,6 +228,19 @@ def resolve_dynamic_choices(question, template=None): elif source_type == 'api_endpoint': choices = _resolve_api_endpoint(dc) elif source_type == 'jinja2': + # Reached only by survey specs stored before the config validation + # below started rejecting this source type. Refuse rather than + # render: the payload is already in the database, and this is the + # call that would execute it. + if not jinja2_source_enabled(): + logger.warning( + 'Refusing to render a jinja2 dynamic_choices template for survey variable %s: ' + 'the jinja2 source type is disabled (%s is not True). The stored survey spec ' + 'predates the validation that now rejects it.', + variable, + SURVEY_DYNAMIC_CHOICES_JINJA2_SETTING, + ) + return [] choices = _resolve_jinja2(dc, template) else: logger.warning('Unknown dynamic_choices source_type: %s', source_type) @@ -68,7 +253,7 @@ def resolve_dynamic_choices(question, template=None): choices = [str(c) for c in choices] # Cache results - if cache_ttl and cache_ttl > 0: + if ck is not None and cache_ttl and cache_ttl > 0: cache.set(ck, choices, timeout=cache_ttl) return choices @@ -143,21 +328,53 @@ def _resolve_api_endpoint(dc): if not url: return [] + refusal = api_destination_refusal(url) + if refusal: + # Logged, not returned: the reason describes the internal network, and + # the caller of the choices endpoint is any user with `start` permission. + logger.warning('Refusing dynamic choices API request to %s: %s', url, refusal) + return [] + method = dc.get('method', 'GET').upper() + if method not in ('GET', 'POST'): + logger.warning('Refusing dynamic choices API request to %s: method %s is not allowed', url, method) + return [] + headers = dc.get('headers', {}) timeout = dc.get('timeout', 10) + try: + timeout = min(float(timeout), API_MAX_TIMEOUT) + except (TypeError, ValueError): + timeout = 10 json_path = dc.get('json_path', '') value_field = dc.get('value_field', '') try: + # allow_redirects=False on purpose. Following them would re-open exactly + # what api_destination_refusal() just closed: the first hop is checked, + # every hop after it is chosen by the peer. + kwargs = dict(headers=headers, timeout=timeout, allow_redirects=False, stream=True) if method == 'POST': - body = dc.get('body', {}) - resp = requests.post(url, json=body, headers=headers, timeout=timeout) + resp = requests.post(url, json=dc.get('body', {}), **kwargs) else: - resp = requests.get(url, headers=headers, timeout=timeout) + resp = requests.get(url, **kwargs) + + with resp: + if resp.is_redirect or resp.is_permanent_redirect: + logger.warning( + 'Refusing dynamic choices API response from %s: redirect to %s not followed', + url, + resp.headers.get('Location', '(no Location)'), + ) + return [] + resp.raise_for_status() - resp.raise_for_status() - data = resp.json() + # Read bounded rather than trusting Content-Length: the peer sets it. + body = resp.raw.read(API_RESPONSE_MAX_BYTES + 1, decode_content=True) + if len(body) > API_RESPONSE_MAX_BYTES: + logger.warning('Refusing dynamic choices API response from %s: larger than %s bytes', url, API_RESPONSE_MAX_BYTES) + return [] + data = json.loads(body) except Exception: logger.exception('Dynamic choices API request failed for %s', url) return [] @@ -177,26 +394,45 @@ def _resolve_api_endpoint(dc): # Extract values if value_field: - return [item.get(value_field, '') for item in data if isinstance(item, dict)] + return [item.get(value_field, '') for item in data if isinstance(item, dict)][:MAX_CHOICES] else: - return [str(item) for item in data] + return [str(item) for item in data][:MAX_CHOICES] def _resolve_jinja2(dc, template=None): """ Resolve choices from a Jinja2 template expression. + Disabled unless ``SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED`` is True in the + server settings file; see the note on that setting above. When it is on, + the template is rendered in a sandbox with the globals removed and the + filter set reduced to ``JINJA2_ALLOWED_FILTERS``. + + Even then the sandbox is a hardening measure, not a trust boundary -- + sandbox escapes are found periodically, and a template that renders in the + web process is worth an escape to whoever plants it. Prefer db_query or + api_endpoint. + Config format: { - "template": "{{ groups | map(attribute='name') | list }}" + "template": "{{ groups | sort | tojson }}" } Available context variables: - hosts: list of host names from the template's inventory - groups: list of group names from the template's inventory """ + if not jinja2_source_enabled(): + logger.warning( + 'Refusing to render a jinja2 dynamic_choices template: the jinja2 source type ' + 'is disabled (%s is not True).', + SURVEY_DYNAMIC_CHOICES_JINJA2_SETTING, + ) + return [] + try: - from jinja2 import Environment, BaseLoader, StrictUndefined + from jinja2 import BaseLoader, StrictUndefined + from jinja2.sandbox import SandboxedEnvironment except ImportError: logger.error('Jinja2 is required for dynamic_choices jinja2 source_type') return [] @@ -216,14 +452,23 @@ def _resolve_jinja2(dc, template=None): context['groups'] = list(Group.objects.filter(inventory_id=inventory_id).values_list('name', flat=True)[:500]) try: - env = Environment(loader=BaseLoader(), undefined=StrictUndefined) + env = SandboxedEnvironment(loader=BaseLoader(), undefined=StrictUndefined) + # Jinja2 seeds every environment with range, dict, lipsum, cycler, + # namespace and joiner. cycler and joiner are class instances, which is + # exactly what `{{ cycler.__init__.__globals__.os }}` walks. The sandbox + # blocks that attribute access; there is still no reason to leave the + # objects reachable. + env.globals.clear() + env.filters = {name: f for name, f in env.filters.items() if name in JINJA2_ALLOWED_FILTERS} + env.tests = {} + tmpl = env.from_string(template_str) result = tmpl.render(**context) # Try to parse as JSON list parsed = json.loads(result) if isinstance(parsed, list): - return parsed + return parsed[:MAX_CHOICES] return [] except Exception: logger.exception('Dynamic choices Jinja2 evaluation failed') @@ -253,6 +498,15 @@ def validate_dynamic_choices_config(dc): errors.append(f"dynamic_choices source_type must be one of: db_query, api_endpoint, jinja2. Got '{source_type}'.") return errors + if source_type == 'jinja2' and not jinja2_source_enabled(): + errors.append( + "dynamic_choices source_type 'jinja2' is disabled: it renders a template supplied " + "with the survey inside the server process. Use db_query or api_endpoint, or ask an " + "administrator to set SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED = True in the server " + "settings file." + ) + return errors + cache_ttl = dc.get('cache_ttl', 60) if not isinstance(cache_ttl, int) or cache_ttl < 0: errors.append("dynamic_choices cache_ttl must be a non-negative integer.") @@ -269,6 +523,17 @@ def validate_dynamic_choices_config(dc): url = dc.get('url', '') if not url or not isinstance(url, str): errors.append("dynamic_choices api_endpoint requires a non-empty 'url' string.") + else: + # Refuse at save time as well as at fetch time, so the editor learns + # the destination is not permitted instead of getting a survey that + # silently resolves to nothing. + refusal = api_destination_refusal(url) + if refusal: + errors.append( + "dynamic_choices api_endpoint url is not permitted: " + f"{refusal}. Destinations are named by an administrator in " + f"{SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST_SETTING}." + ) elif source_type == 'jinja2': tmpl = dc.get('template', '') @@ -276,3 +541,49 @@ def validate_dynamic_choices_config(dc): errors.append("dynamic_choices jinja2 requires a non-empty 'template' string.") return errors + + +def survey_choice_list(question, template=None): + """ + The permitted values for a choice question, and whether they are dynamic. + + For a dynamic question this resolves the list at the moment it is asked for + -- which, on the launch path, is the point of the exercise. Validation there + used to be skipped with the comment "validated at resolve time", but the + resolve endpoint only hands options to the UI; nothing on the launch path + ever compared the submitted value against them, so a direct API client could + pass any extra_var for a question that presents as a fixed dropdown. + + A failed resolution yields an empty list, and an empty list rejects every + answer. That is deliberate: if the permitted values cannot be determined, the + request cannot be validated, and accepting it would mean trusting the caller + for exactly the field this check constrains. The choices could not have been + offered in the UI either. + """ + dc = question.get('dynamic_choices') or {} + if dc.get('enabled'): + try: + resolved = resolve_dynamic_choices(question, template=template) + except Exception: + logger.exception('Failed to resolve dynamic choices for survey variable %s', question.get('variable')) + resolved = None + return list(resolved or []), True + + choices = question.get('choices', []) + if isinstance(choices, str): + choices = [choice for choice in choices.splitlines() if choice.strip() != ''] + return list(choices), False + + +def choice_error_message(question, value, choices, is_dynamic): + """The message for a value that is not among a question's permitted ones.""" + variable = question.get('variable') + if not is_dynamic: + return "Value %s for '%s' expected to be one of %s." % (value, variable, choices) + if not choices: + # Distinguish "resolved to nothing" from "not in the list". The first is + # usually a broken source, and reporting it as a bad answer sends the + # operator looking in the wrong place. + return "Value %s for '%s' could not be validated: its dynamic choices resolved to no options." % (value, variable) + # The list can hold up to MAX_CHOICES entries, so it is counted, not printed. + return "Value %s for '%s' expected to be one of its %s dynamic choices." % (value, variable, len(choices)) diff --git a/forail/settings/defaults/forail_settings.py b/forail/settings/defaults/forail_settings.py index 0b11d3d..4ba4e7f 100644 --- a/forail/settings/defaults/forail_settings.py +++ b/forail/settings/defaults/forail_settings.py @@ -153,3 +153,42 @@ }, } } + + +# ----------------------------- +# -- Dynamic survey choices -- +# ----------------------------- +# A survey question may resolve its choices from the database, from an external +# API, or by rendering a Jinja2 template. The last one executes a template that +# a job-template editor supplies, inside the web process, whenever a user with +# `start` permission opens the launch prompt -- which is a code-execution path, +# not a formatting convenience. +# +# It is therefore off, and deliberately NOT a database-backed setting: leaving +# it out of /api/v2/settings/ means the API surface used to store a payload +# cannot also be used to enable its execution. Turning it on takes a change to +# a settings file on the server. +# +# When on, templates render in a Jinja2 sandbox with globals removed and a +# reduced filter set (see forail/main/services/dynamic_survey.py). Treat that as +# hardening rather than a boundary: prefer the db_query or api_endpoint source +# types, which need no code execution at all. +SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED = False + + +# The `api_endpoint` source makes the server fetch a URL taken from the survey. +# Without a destination policy that is an SSRF primitive: whoever edits a job +# template chooses the address and the server reaches it from inside the +# cluster, then hands the JSON back to any user with `start` permission. +# +# So destinations are named here, by an operator, and matched exactly -- no +# wildcards, no suffix matching. Empty (the default) disables the source type. +# https only; redirects are never followed; the response is read bounded. +# +# SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST = ['cmdb.internal.example.com'] +# +# Private addresses are allowed on purpose -- an on-prem CMDB is the ordinary +# case, and naming the host here is the trust decision. Loopback, link-local +# (cloud metadata), multicast and reserved addresses are refused even for a +# listed host, so a hijacked DNS record cannot redirect the fetch inward. +SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST = [] diff --git a/tests_standalone/test_dynamic_survey_standalone.py b/tests_standalone/test_dynamic_survey_standalone.py index 38005a1..1dd1208 100644 --- a/tests_standalone/test_dynamic_survey_standalone.py +++ b/tests_standalone/test_dynamic_survey_standalone.py @@ -7,12 +7,18 @@ import os from unittest.mock import patch, MagicMock, PropertyMock -# Mock Django modules before importing our code +# Mock Django modules before importing our code. +# +# django.db has to be stubbed alongside the rest: forail/__init__.py does +# `from django.db import connection` at import time, and a bare MagicMock under +# 'django' is not a package, so that line is what used to make this file +# uncollectable on its own. It was excluded from CI for it. sys.modules['django'] = MagicMock() sys.modules['django.apps'] = MagicMock() sys.modules['django.core'] = MagicMock() sys.modules['django.core.cache'] = MagicMock() sys.modules['django.conf'] = MagicMock() +sys.modules['django.db'] = MagicMock() import pytest @@ -23,7 +29,10 @@ if 'forail.main.services.dynamic_survey' in sys.modules: del sys.modules['forail.main.services.dynamic_survey'] +from forail.main.services import dynamic_survey from forail.main.services.dynamic_survey import ( + survey_choice_list, + choice_error_message, validate_dynamic_choices_config, _resolve_api_endpoint, _resolve_jinja2, @@ -34,6 +43,68 @@ ) +class _Settings: + """Stands in for django.conf.settings. + + The real settings object is a MagicMock here, and every attribute of a + MagicMock is truthy -- which would silently enable the jinja2 source type + in every test. The service reads the flag with `is True` for that reason; + this class lets a test say which value it wants. + """ + + def __init__(self, **attrs): + self.__dict__.update(attrs) + + +def jinja2_enabled(value=True): + """Patch the service's settings so the jinja2 source type is on/off.""" + return patch.object( + dynamic_survey, 'settings', _Settings(SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED=value) + ) + + +def api_allowed(*hosts): + """Patch the service's settings so those hosts are permitted destinations.""" + return patch.object( + dynamic_survey, 'settings', _Settings(SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST=list(hosts)) + ) + + +def resolves_to(address='93.184.216.34'): + """Patch name resolution, so no test depends on a real DNS answer.""" + family = 2 if ':' not in address else 10 + return patch.object( + dynamic_survey, + 'socket', + MagicMock( + IPPROTO_TCP=6, + getaddrinfo=MagicMock(return_value=[(family, 1, 6, '', (address, 443))]), + ), + ) + + +def fake_template(pk=1, organization_id=1, inventory_id=1): + """A job template, as the cache key reads it.""" + t = MagicMock() + t.pk = pk + t.organization_id = organization_id + t.inventory_id = inventory_id + return t + + +def api_response(payload, status_ok=True): + """A stand-in for requests' Response, as the fetch path actually uses it.""" + resp = MagicMock() + resp.is_redirect = False + resp.is_permanent_redirect = False + resp.headers = {} + resp.raw.read.return_value = json.dumps(payload).encode() + resp.raise_for_status = MagicMock() if status_ok else MagicMock(side_effect=Exception('http error')) + resp.__enter__ = MagicMock(return_value=resp) + resp.__exit__ = MagicMock(return_value=False) + return resp + + # ===== validate_dynamic_choices_config ===== class TestValidation: @@ -44,11 +115,36 @@ def test_valid_db_query(self): def test_valid_api_endpoint(self): dc = {'enabled': True, 'source_type': 'api_endpoint', 'url': 'https://example.com/api', 'cache_ttl': 30} - assert validate_dynamic_choices_config(dc) == [] + with api_allowed('example.com'), resolves_to(): + assert validate_dynamic_choices_config(dc) == [] + + def test_api_endpoint_rejected_without_an_allowlist(self): + # Same posture as the jinja2 source type: nothing is reachable until an + # operator names it. + dc = {'enabled': True, 'source_type': 'api_endpoint', 'url': 'https://example.com/api', 'cache_ttl': 30} + errors = validate_dynamic_choices_config(dc) + assert any('SURVEY_DYNAMIC_CHOICES_API_ALLOWLIST' in e for e in errors) - def test_valid_jinja2(self): + def test_jinja2_rejected_by_default(self): + # The source type executes a template in the web process, so a survey + # spec may not even be saved with it unless an operator opted in. dc = {'enabled': True, 'source_type': 'jinja2', 'template': '{{ hosts }}', 'cache_ttl': 10} - assert validate_dynamic_choices_config(dc) == [] + errors = validate_dynamic_choices_config(dc) + assert len(errors) == 1 + assert 'SURVEY_DYNAMIC_CHOICES_JINJA2_ENABLED' in errors[0] + + def test_valid_jinja2_when_operator_enabled(self): + dc = {'enabled': True, 'source_type': 'jinja2', 'template': '{{ hosts }}', 'cache_ttl': 10} + with jinja2_enabled(): + assert validate_dynamic_choices_config(dc) == [] + + def test_jinja2_not_enabled_by_a_truthy_value(self): + # A stray "true"/1 in a settings file must not turn code execution on. + dc = {'enabled': True, 'source_type': 'jinja2', 'template': '{{ hosts }}'} + for value in ('True', 'true', 1, [1]): + with jinja2_enabled(value): + errors = validate_dynamic_choices_config(dc) + assert errors, f'{value!r} should not enable the jinja2 source type' def test_disabled_always_valid(self): assert validate_dynamic_choices_config({'enabled': False}) == [] @@ -85,7 +181,8 @@ def test_api_missing_url(self): def test_jinja2_missing_template(self): dc = {'enabled': True, 'source_type': 'jinja2'} - errors = validate_dynamic_choices_config(dc) + with jinja2_enabled(): + errors = validate_dynamic_choices_config(dc) assert any("template" in e for e in errors) def test_negative_ttl(self): @@ -109,45 +206,34 @@ def test_all_allowed_fields(self): # ===== _resolve_api_endpoint ===== class TestApiEndpoint: + """The fetch path, with the destination named by an operator.""" @patch('forail.main.services.dynamic_survey.requests') def test_simple_list(self, mock_req): - resp = MagicMock() - resp.json.return_value = ['a', 'b', 'c'] - resp.raise_for_status = MagicMock() - mock_req.get.return_value = resp - - result = _resolve_api_endpoint({'url': 'https://example.com/list'}) + mock_req.get.return_value = api_response(['a', 'b', 'c']) + with api_allowed('example.com'), resolves_to(): + result = _resolve_api_endpoint({'url': 'https://example.com/list'}) assert result == ['a', 'b', 'c'] @patch('forail.main.services.dynamic_survey.requests') def test_json_path(self, mock_req): - resp = MagicMock() - resp.json.return_value = {'data': {'items': ['x', 'y']}} - resp.raise_for_status = MagicMock() - mock_req.get.return_value = resp - - result = _resolve_api_endpoint({'url': 'https://example.com', 'json_path': 'data.items'}) + mock_req.get.return_value = api_response({'data': {'items': ['x', 'y']}}) + with api_allowed('example.com'), resolves_to(): + result = _resolve_api_endpoint({'url': 'https://example.com', 'json_path': 'data.items'}) assert result == ['x', 'y'] @patch('forail.main.services.dynamic_survey.requests') def test_value_field(self, mock_req): - resp = MagicMock() - resp.json.return_value = [{'name': 'srv1'}, {'name': 'srv2'}] - resp.raise_for_status = MagicMock() - mock_req.get.return_value = resp - - result = _resolve_api_endpoint({'url': 'https://example.com', 'value_field': 'name'}) + mock_req.get.return_value = api_response([{'name': 'srv1'}, {'name': 'srv2'}]) + with api_allowed('example.com'), resolves_to(): + result = _resolve_api_endpoint({'url': 'https://example.com', 'value_field': 'name'}) assert result == ['srv1', 'srv2'] @patch('forail.main.services.dynamic_survey.requests') def test_post_method(self, mock_req): - resp = MagicMock() - resp.json.return_value = ['p1', 'p2'] - resp.raise_for_status = MagicMock() - mock_req.post.return_value = resp - - result = _resolve_api_endpoint({'url': 'https://example.com', 'method': 'POST', 'body': {}}) + mock_req.post.return_value = api_response(['p1', 'p2']) + with api_allowed('example.com'), resolves_to(): + result = _resolve_api_endpoint({'url': 'https://example.com', 'method': 'POST', 'body': {}}) assert result == ['p1', 'p2'] mock_req.post.assert_called_once() @@ -157,38 +243,267 @@ def test_empty_url(self): @patch('forail.main.services.dynamic_survey.requests') def test_error_returns_empty(self, mock_req): mock_req.get.side_effect = Exception("fail") - assert _resolve_api_endpoint({'url': 'https://bad.example.com'}) == [] + with api_allowed('bad.example.com'), resolves_to(): + assert _resolve_api_endpoint({'url': 'https://bad.example.com'}) == [] @patch('forail.main.services.dynamic_survey.requests') def test_non_list_response(self, mock_req): - resp = MagicMock() - resp.json.return_value = {"not": "a list"} - resp.raise_for_status = MagicMock() + mock_req.get.return_value = api_response({"not": "a list"}) + with api_allowed('example.com'), resolves_to(): + result = _resolve_api_endpoint({'url': 'https://example.com'}) + assert result == [] + + @patch('forail.main.services.dynamic_survey.requests') + def test_redirect_is_not_followed(self, mock_req): + # The first hop is what the allowlist checked; every hop after it is + # chosen by the peer, so following one re-opens the SSRF. + resp = api_response([]) + resp.is_redirect = True + resp.headers = {'Location': 'http://169.254.169.254/latest/meta-data/'} mock_req.get.return_value = resp + with api_allowed('example.com'), resolves_to(): + assert _resolve_api_endpoint({'url': 'https://example.com'}) == [] + assert mock_req.get.call_args[1]['allow_redirects'] is False + + @patch('forail.main.services.dynamic_survey.requests') + def test_oversized_response_is_refused(self, mock_req): + resp = api_response([]) + resp.raw.read.return_value = b'x' * (dynamic_survey.API_RESPONSE_MAX_BYTES + 1) + mock_req.get.return_value = resp + with api_allowed('example.com'), resolves_to(): + assert _resolve_api_endpoint({'url': 'https://example.com'}) == [] + + @patch('forail.main.services.dynamic_survey.requests') + def test_timeout_is_capped(self, mock_req): + mock_req.get.return_value = api_response([]) + with api_allowed('example.com'), resolves_to(): + _resolve_api_endpoint({'url': 'https://example.com', 'timeout': 9999}) + assert mock_req.get.call_args[1]['timeout'] == dynamic_survey.API_MAX_TIMEOUT - result = _resolve_api_endpoint({'url': 'https://example.com'}) - assert result == [] + +class TestApiDestinationPolicy: + """ + Destination control, which is what turns this source type from an SSRF + primitive into a fetch from somewhere an operator named. + """ + + @patch('forail.main.services.dynamic_survey.requests') + def test_no_allowlist_refuses_everything(self, mock_req): + # The default posture: the source type is off until an operator names a + # destination. + assert _resolve_api_endpoint({'url': 'https://example.com'}) == [] + mock_req.get.assert_not_called() + + @patch('forail.main.services.dynamic_survey.requests') + def test_unlisted_host_is_refused(self, mock_req): + with api_allowed('cmdb.internal.example.com'), resolves_to(): + assert _resolve_api_endpoint({'url': 'https://evil.example.net/x'}) == [] + mock_req.get.assert_not_called() + + @patch('forail.main.services.dynamic_survey.requests') + def test_http_is_refused(self, mock_req): + with api_allowed('example.com'), resolves_to(): + assert _resolve_api_endpoint({'url': 'http://example.com'}) == [] + mock_req.get.assert_not_called() + + @patch('forail.main.services.dynamic_survey.requests') + def test_listed_host_resolving_to_metadata_is_refused(self, mock_req): + # The case the allowlist alone does not cover: a listed name whose DNS + # answer points at the cloud metadata service. + with api_allowed('cmdb.internal.example.com'), resolves_to('169.254.169.254'): + assert _resolve_api_endpoint({'url': 'https://cmdb.internal.example.com/x'}) == [] + mock_req.get.assert_not_called() + + @patch('forail.main.services.dynamic_survey.requests') + def test_listed_host_resolving_to_loopback_is_refused(self, mock_req): + with api_allowed('cmdb.internal.example.com'), resolves_to('127.0.0.1'): + assert _resolve_api_endpoint({'url': 'https://cmdb.internal.example.com/x'}) == [] + mock_req.get.assert_not_called() + + @patch('forail.main.services.dynamic_survey.requests') + def test_private_address_is_allowed(self, mock_req): + # An on-prem CMDB on RFC1918 is the ordinary use of this feature; the + # operator naming the host is the trust decision. + mock_req.get.return_value = api_response(['srv1']) + with api_allowed('cmdb.internal.example.com'), resolves_to('10.4.1.7'): + assert _resolve_api_endpoint({'url': 'https://cmdb.internal.example.com/x'}) == ['srv1'] + + @patch('forail.main.services.dynamic_survey.requests') + def test_non_get_post_method_is_refused(self, mock_req): + with api_allowed('example.com'), resolves_to(): + assert _resolve_api_endpoint({'url': 'https://example.com', 'method': 'DELETE'}) == [] + + def test_validation_refuses_an_unlisted_destination(self): + dc = {'enabled': True, 'source_type': 'api_endpoint', 'url': 'https://evil.example.net'} + with api_allowed('cmdb.internal.example.com'), resolves_to(): + errors = validate_dynamic_choices_config(dc) + assert any('not permitted' in e for e in errors) + + def test_validation_accepts_a_listed_destination(self): + dc = {'enabled': True, 'source_type': 'api_endpoint', 'url': 'https://cmdb.internal.example.com/x'} + with api_allowed('cmdb.internal.example.com'), resolves_to('10.4.1.7'): + assert validate_dynamic_choices_config(dc) == [] # ===== _resolve_jinja2 ===== class TestJinja2: + """The renderer, with the source type turned on by an operator.""" def test_static_list(self): - result = _resolve_jinja2({'template': '["opt1", "opt2"]'}) + with jinja2_enabled(): + result = _resolve_jinja2({'template': '["opt1", "opt2"]'}) assert result == ['opt1', 'opt2'] def test_empty_template(self): - assert _resolve_jinja2({'template': ''}) == [] + with jinja2_enabled(): + assert _resolve_jinja2({'template': ''}) == [] def test_invalid_output(self): # Non-JSON result - result = _resolve_jinja2({'template': 'not json'}) + with jinja2_enabled(): + result = _resolve_jinja2({'template': 'not json'}) + assert result == [] + + def test_filters_still_work(self): + with jinja2_enabled(): + result = _resolve_jinja2({'template': '{{ ["b", "a", "b"] | unique | sort | list | tojson }}'}) + assert result == ['a', 'b'] + + def test_result_is_capped(self): + with jinja2_enabled(): + result = _resolve_jinja2({'template': '{{ (["x"] * 600) | list | tojson }}'}) + assert len(result) == 500 + + +class TestJinja2Disabled: + """Default posture: the template is never rendered at all.""" + + def test_renderer_refuses(self): + # A template that would raise if it were rendered proves the renderer + # was never reached, not merely that the output was discarded. + assert _resolve_jinja2({'template': '["ran"]'}) == [] + + @patch('forail.main.services.dynamic_survey.cache') + @patch('forail.main.services.dynamic_survey._resolve_jinja2') + def test_dispatch_does_not_call_the_renderer(self, mock_jinja, mock_cache): + # Survey specs saved before the source type was refused still sit in the + # database; resolving one must not execute it. + mock_cache.get.return_value = None + q = { + 'variable': 'v', + 'dynamic_choices': { + 'enabled': True, + 'source_type': 'jinja2', + 'template': '{{ cycler.__init__.__globals__ }}', + 'cache_ttl': 10, + }, + } + assert resolve_dynamic_choices(q) == [] + mock_jinja.assert_not_called() + + @patch('forail.main.services.dynamic_survey.cache') + @patch('forail.main.services.dynamic_survey._resolve_jinja2') + def test_refusal_is_not_cached(self, mock_jinja, mock_cache): + # Caching the empty result would mask the warning for the whole TTL. + mock_cache.get.return_value = None + q = {'variable': 'v', 'dynamic_choices': {'enabled': True, 'source_type': 'jinja2', 'cache_ttl': 300}} + resolve_dynamic_choices(q) + mock_cache.set.assert_not_called() + + +# Known template-to-Python routes. These are the expressions the original +# unsandboxed Environment answered: `{{ cycler.__init__.__globals__.os.name }}` +# returned `posix`, which is a read from the module table and one attribute away +# from `os.system`. +JINJA2_ESCAPES = [ + "{{ cycler.__init__.__globals__.os.name }}", + "{{ cycler.__init__.__globals__['os'].name }}", + "{{ joiner.__init__.__globals__ }}", + "{{ namespace.__init__.__globals__ }}", + "{{ ''.__class__.__mro__[1].__subclasses__() }}", + "{{ ''.__class__.__base__.__subclasses__() }}", + "{{ [].__class__.__base__.__subclasses__() }}", + "{{ lipsum.__globals__ }}", + "{{ self._TemplateReference__context }}", + "{{ config }}", + "{{ request }}", + "{{ ''.__class__.__mro__[1].__subclasses__()[0].__init__.__globals__ }}", +] + + +class TestJinja2SandboxEscapes: + """The escapes above, asserted against the operator-enabled path.""" + + @pytest.mark.parametrize('expression', JINJA2_ESCAPES) + def test_escape_yields_no_choices(self, expression): + with jinja2_enabled(): + assert _resolve_jinja2({'template': expression}) == [] + + @pytest.mark.parametrize('expression', JINJA2_ESCAPES) + def test_escape_never_reaches_the_module_table(self, expression): + # Wrapping the expression in a JSON list matters: rendered bare, an + # escape returns a Python repr that fails json.loads, so the empty + # result would prove nothing. Wrapped, a successful escape parses + # cleanly and comes back as a populated list. Against the pre-fix + # renderer this exact shape returned ['x', 'posix'] for the cycler + # payload, and the PATH environment variable for its os.environ variant. + with jinja2_enabled(): + result = _resolve_jinja2({'template': '["x", "' + expression + '"]'}) assert result == [] - def test_expression_eval(self): - result = _resolve_jinja2({'template': '{{ range(1,4) | list | tojson }}'}) - assert result == [1, 2, 3] + def test_globals_are_gone(self): + # range/dict/lipsum/cycler/namespace/joiner are removed outright, so the + # escape above has nothing to start from even before the sandbox runs. + with jinja2_enabled(): + for name in ('range', 'dict', 'lipsum', 'cycler', 'namespace', 'joiner'): + assert _resolve_jinja2({'template': f'{{{{ {name} }}}}'}) == [] + + def test_disallowed_filter_is_gone(self): + with jinja2_enabled(): + assert _resolve_jinja2({'template': "{{ ['a'] | pprint }}"}) == [] + + +class TestCacheScope: + """ + The key has to carry the scope the answer was resolved in, or one tenant is + served another's host names. + """ + + CONFIG = {'enabled': True, 'source_type': 'db_query', 'model': 'hosts', 'cache_ttl': 60} + + def test_same_question_different_inventory_gets_a_different_key(self): + a = dynamic_survey._cache_key('v', self.CONFIG, fake_template(pk=1, inventory_id=10)) + b = dynamic_survey._cache_key('v', self.CONFIG, fake_template(pk=1, inventory_id=20)) + assert a and b and a != b + + def test_same_question_different_organization_gets_a_different_key(self): + a = dynamic_survey._cache_key('v', self.CONFIG, fake_template(pk=1, organization_id=1)) + b = dynamic_survey._cache_key('v', self.CONFIG, fake_template(pk=1, organization_id=2)) + assert a and b and a != b + + def test_same_question_different_template_gets_a_different_key(self): + a = dynamic_survey._cache_key('v', self.CONFIG, fake_template(pk=1)) + b = dynamic_survey._cache_key('v', self.CONFIG, fake_template(pk=2)) + assert a and b and a != b + + def test_identical_scope_reuses_the_key(self): + a = dynamic_survey._cache_key('v', self.CONFIG, fake_template()) + b = dynamic_survey._cache_key('v', self.CONFIG, fake_template()) + assert a == b + + def test_no_template_means_no_key(self): + assert dynamic_survey._cache_key('v', self.CONFIG, None) is None + + @patch('forail.main.services.dynamic_survey.cache') + @patch('forail.main.services.dynamic_survey._resolve_db_query') + def test_unscoped_resolution_is_not_cached(self, mock_resolve, mock_cache): + # Better to resolve every time than to write an entry every tenant reads. + mock_cache.get.return_value = None + mock_resolve.return_value = ['h1'] + q = {'variable': 'v', 'dynamic_choices': dict(self.CONFIG)} + assert resolve_dynamic_choices(q) == ['h1'] + mock_cache.set.assert_not_called() # ===== _resolve_db_query ===== @@ -252,7 +567,7 @@ def test_no_dc_returns_none(self, mock_cache): def test_cache_hit(self, mock_resolve, mock_cache): mock_cache.get.return_value = ['c1', 'c2'] q = {'variable': 'v', 'dynamic_choices': {'enabled': True, 'source_type': 'db_query', 'model': 'hosts', 'cache_ttl': 60}} - result = resolve_dynamic_choices(q) + result = resolve_dynamic_choices(q, template=fake_template()) assert result == ['c1', 'c2'] mock_resolve.assert_not_called() @@ -262,7 +577,7 @@ def test_cache_miss(self, mock_resolve, mock_cache): mock_cache.get.return_value = None mock_resolve.return_value = ['h1', 'h2'] q = {'variable': 'v', 'dynamic_choices': {'enabled': True, 'source_type': 'db_query', 'model': 'hosts', 'cache_ttl': 120}} - result = resolve_dynamic_choices(q) + result = resolve_dynamic_choices(q, template=fake_template()) assert result == ['h1', 'h2'] mock_cache.set.assert_called_once() # Verify TTL is passed @@ -283,7 +598,8 @@ def test_jinja2_source(self, mock_resolve, mock_cache): mock_cache.get.return_value = None mock_resolve.return_value = ['j1', 'j2'] q = {'variable': 'v', 'dynamic_choices': {'enabled': True, 'source_type': 'jinja2', 'template': '{{ x }}', 'cache_ttl': 10}} - result = resolve_dynamic_choices(q) + with jinja2_enabled(): + result = resolve_dynamic_choices(q) assert result == ['j1', 'j2'] @patch('forail.main.services.dynamic_survey.cache') @@ -305,3 +621,66 @@ def test_results_are_stringified(self, mock_resolve, mock_cache): if __name__ == '__main__': pytest.main([__file__, '-v']) + + +# ===== launch-time validation of dynamic answers ===== + +class TestSurveyChoiceList: + """ + What the launch path compares a submitted answer against. + + Before this existed the dynamic branch was `pass`, with a comment saying the + values were checked at resolve time. They were not: the resolve endpoint only + hands options to the UI, so a direct API client could send any value for a + question that looks like a fixed dropdown. + """ + + STATIC = {'variable': 'env', 'choices': 'dev\nstaging\nprod\n'} + DYNAMIC = {'variable': 'host', 'dynamic_choices': {'enabled': True, 'source_type': 'db_query', 'model': 'hosts'}} + + def test_static_choices_are_split(self): + choices, is_dynamic = survey_choice_list(self.STATIC) + assert choices == ['dev', 'staging', 'prod'] + assert is_dynamic is False + + def test_static_choices_as_a_list_pass_through(self): + choices, is_dynamic = survey_choice_list({'variable': 'x', 'choices': ['a', 'b']}) + assert (choices, is_dynamic) == (['a', 'b'], False) + + @patch('forail.main.services.dynamic_survey.resolve_dynamic_choices') + def test_dynamic_choices_are_resolved(self, mock_resolve): + mock_resolve.return_value = ['web-01', 'web-02'] + choices, is_dynamic = survey_choice_list(self.DYNAMIC, template=fake_template()) + assert (choices, is_dynamic) == (['web-01', 'web-02'], True) + assert mock_resolve.call_args[1]['template'] is not None + + @patch('forail.main.services.dynamic_survey.resolve_dynamic_choices') + def test_a_failed_resolution_permits_nothing(self, mock_resolve): + # Fail closed: an unresolvable list means the answer cannot be checked, + # and accepting it would trust the caller for the one field this exists + # to constrain. + mock_resolve.side_effect = Exception('source down') + choices, is_dynamic = survey_choice_list(self.DYNAMIC, template=fake_template()) + assert (choices, is_dynamic) == ([], True) + + @patch('forail.main.services.dynamic_survey.resolve_dynamic_choices') + def test_none_resolution_permits_nothing(self, mock_resolve): + mock_resolve.return_value = None + assert survey_choice_list(self.DYNAMIC, template=fake_template()) == ([], True) + + +class TestChoiceErrorMessage: + + def test_static_lists_the_options(self): + msg = choice_error_message({'variable': 'env'}, 'x', ['dev', 'prod'], False) + assert "'env'" in msg and 'dev' in msg + + def test_dynamic_counts_rather_than_prints(self): + # The list can hold up to MAX_CHOICES entries. + msg = choice_error_message({'variable': 'host'}, 'x', [f'h{i}' for i in range(500)], True) + assert '500 dynamic choices' in msg + assert 'h1' not in msg + + def test_empty_dynamic_list_says_so(self): + msg = choice_error_message({'variable': 'host'}, 'x', [], True) + assert 'resolved to no options' in msg