diff --git a/CHANGELOG.md b/CHANGELOG.md index eb5b2e1..668ef15 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.7.0] - 2026-08-04 + +### Added + +- `StaticTokenAuth` and a new `auth=` parameter on `AsyncBCClient`, for embedders + that already hold a Business Central access token. Pass a token string or a + callable — a callable is re-invoked per request, so a long-running process + picks up a refreshed token instead of pinning one that expires. Injected auth + bypasses the profile's `auth_method` entirely, which is what lets a `browser` + profile work somewhere that has no browser and no loopback listener to bind. +- `bcli.queries`, a reusable saved-query module extracted from the CLI: catalog + loading, parameter validation against each parameter's declared + type/pattern/min/max/enum, and `${{ params.X }}` resolution with OData + escaping. `QueryCatalogError`, `QueryError` and `QueryParamError` are exported + from the package root. `bcli q` is now a consumer of this module; its output, + errors and exit codes are unchanged. +- Python 3.14 in the supported-versions classifiers. `requires-python` stays at + `>=3.11`. + +### Fixed + +- `build_url` now validates `entity_set_name` and `record_id` as single URL path + components, rejecting raw `/`, `\`, `?`, `#` and the `.`/`..` segments. Both + values are spliced directly into the request path, so a record key containing + a path separator could compose a URL addressing a different entity than the + one the endpoint registry was consulted about — the registry lookup and the + `disable_standard_api` gate both key on the entity-set name alone. Keys that + legitimately contain quotes, commas, equals signs, hyphens or parentheses + inside a quoted string are unaffected; percent-encode a separator that is + genuinely part of a key. The same validation applies to the key inside a + bound-action invocation, whose resolver checks the registry for the parent + entity set only. +- `test_no_context_policy_path` no longer asserts against the developer's real + `~/.config/bcli`, so it passes on a machine that has recorded a bcli error + rather than only on a clean CI home. + ## [0.6.2] - 2026-07-22 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index 2deace0..f633fcd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ build-backend = "hatchling.build" # installed CLI binary (`bcli`) are unaffected — only `pip install` / # `uv tool install` use this name. name = "bc-cli" -version = "0.6.2" +version = "0.7.0" description = "Python SDK and CLI for Microsoft Dynamics 365 Business Central APIs" readme = "README.md" license = "Apache-2.0" @@ -26,6 +26,7 @@ classifiers = [ "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", "Topic :: Software Development :: Libraries :: Python Modules", "Topic :: Office/Business :: Financial :: Accounting", ] diff --git a/src/bcli/__init__.py b/src/bcli/__init__.py index 37cbf3a..d5d0c2a 100644 --- a/src/bcli/__init__.py +++ b/src/bcli/__init__.py @@ -1,6 +1,7 @@ """bcli — Python SDK for Microsoft Dynamics 365 Business Central APIs.""" from bcli._version import __version__ +from bcli.auth._static import StaticTokenAuth from bcli.client import AsyncBCClient, BCClient from bcli.client._safety import DomainRule, SafeContext from bcli.config import BCConfig, load_config @@ -19,6 +20,7 @@ WorkflowError, ) from bcli.odata import Query +from bcli.queries import QueryCatalogError, QueryError, QueryParamError from bcli.registry import EndpointRegistry __all__ = [ @@ -35,10 +37,14 @@ "ForbiddenError", "NotFoundError", "Query", + "QueryCatalogError", + "QueryError", + "QueryParamError", "RegistryError", "SafeContext", "SafetyError", "ServerError", + "StaticTokenAuth", "ThrottledError", "ValidationError", "WorkflowError", diff --git a/src/bcli/_url.py b/src/bcli/_url.py index 514a244..f4d4567 100644 --- a/src/bcli/_url.py +++ b/src/bcli/_url.py @@ -30,6 +30,43 @@ def _validate_route_segment(name: str, value: str) -> None: ) +#: Characters that let a value escape the URL component it is spliced into. +#: A record key legitimately contains quotes, commas, equals signs, hyphens and +#: even parentheses inside a quoted string (``'ACME (US)'``) — none of which can +#: start a new path segment. These four can. +_URL_STRUCTURE_CHARS = ("/", "\\", "?", "#") + + +def validate_record_key(name: str, value: str) -> None: + """Validate an OData key or entity-set name as a single URL path component. + + ``entity_set_name`` and ``record_id`` are spliced straight into the path, and + a key is caller-influenced anywhere one is accepted from outside. A raw ``/`` + starts a new path segment, so ``foo(1)/../../bar('X'`` composes a URL that + addresses ``bar`` while every earlier check only ever saw ``foo`` — including + the endpoint-registry lookup and the ``disable_standard_api`` gate, both of + which key on the entity-set name alone. + + Raises ValueError on empty input, raw path/query delimiters, and the + ``.``/``..`` traversal segments. + """ + if not value or not value.strip(): + raise ValueError(f"Invalid {name}: must not be empty.") + + for char in _URL_STRUCTURE_CHARS: + if char in value: + raise ValueError( + f"Invalid {name} {value!r}: must not contain {char!r}. A key is a " + f"single URL path component — percent-encode the character if it " + f"is genuinely part of the key." + ) + + if value.strip() in (".", ".."): + raise ValueError( + f"Invalid {name} {value!r}: '.' and '..' are path-traversal segments." + ) + + def build_url( *, environment: str, @@ -49,6 +86,10 @@ def build_url( Custom API: https://api.businesscentral.dynamics.com/v2.0/{env}/api/{pub}/{grp}/{ver}/companies({id})/{entity} """ + validate_record_key("entity_set_name", entity_set_name) + if record_id: + validate_record_key("record_id", record_id) + if publisher and group and version: _validate_route_segment("publisher", publisher) _validate_route_segment("group", group) diff --git a/src/bcli/auth/_static.py b/src/bcli/auth/_static.py new file mode 100644 index 0000000..70d8720 --- /dev/null +++ b/src/bcli/auth/_static.py @@ -0,0 +1,82 @@ +"""Auth provider for an access token the caller already holds. + +Every other provider in this package *acquires* a token — by opening a browser, +by polling a device-code flow, or by exchanging a client secret. Some embedders +have a token already and only need the SDK to use it: + +* a hosted service that performs an on-behalf-of exchange per request, so the BC + call runs as the signed-in user rather than as the service; +* a CI job handed a token by its platform; +* a script or notebook pasting one in for a one-off. + +None of those can use the browser flow, which opens a browser and binds a +loopback listener on whichever machine runs the process. + +Pass either a fixed string or a supplier. Prefer a **supplier** in any +long-running process: it is re-invoked on every call, so a refreshed token is +picked up instead of a pinned one expiring mid-session. Nothing is cached here — +the caller owns the token's lifetime, which is why :meth:`clear_cache` is a +no-op. +""" + +from __future__ import annotations + +import inspect +from collections.abc import Awaitable, Callable + +from bcli.errors import ConfigError + +TokenSupplier = Callable[[], "str | Awaitable[str]"] + + +class StaticTokenAuth: + """Supply a pre-acquired bearer token. Satisfies the ``AuthProvider`` protocol. + + Args: + token: The access token, or a callable returning one (sync or async). + """ + + def __init__(self, token: str | TokenSupplier) -> None: + if callable(token): + self._supplier: TokenSupplier | None = token + self._token: str | None = None + else: + self._supplier = None + self._token = self._validated(token) + + @staticmethod + def _validated(raw: object) -> str: + """Reject the two mistakes that would otherwise surface as a puzzling 401.""" + if not isinstance(raw, str): + raise ConfigError( + f"Access token must be a string, got {type(raw).__name__}. " + "A token supplier must return the raw token." + ) + token = raw.strip() + if not token: + raise ConfigError( + "Access token is empty. Pass the raw bearer token, or a callable " + "that returns one." + ) + if token.lower().startswith("bearer "): + raise ConfigError( + "Access token must not include the 'Bearer ' prefix — the transport " + "adds it, so this would send 'Authorization: Bearer Bearer ...'." + ) + return token + + async def get_access_token(self) -> str: + """Return the token, re-invoking the supplier if one was given.""" + if self._supplier is None: + # Validated in __init__; narrowing for type checkers. + assert self._token is not None + return self._token + + supplied = self._supplier() + if inspect.isawaitable(supplied): + supplied = await supplied + return self._validated(supplied) + + def clear_cache(self) -> None: + """No-op — nothing is cached. Supply a callable if you need invalidation.""" + return None diff --git a/src/bcli/client/_async.py b/src/bcli/client/_async.py index ccfb938..3cde1b5 100644 --- a/src/bcli/client/_async.py +++ b/src/bcli/client/_async.py @@ -6,7 +6,8 @@ from pathlib import Path from typing import Any -from bcli._url import build_companies_url, build_url +from bcli._url import build_companies_url, build_url, validate_record_key +from bcli.auth._base import AuthProvider from bcli.auth._credentials import ClientCredentialsAuth from bcli.client._safety import SafeContext from bcli.client._transport import BCTransport @@ -59,7 +60,15 @@ def _parse_bound_action(entity_set_name: str) -> tuple[str, str, str] | None: m = _BOUND_ACTION_RE.match(entity_set_name) if m is None: return None - return m.group("entity"), m.group("key"), m.group("qualified") + entity, key, qualified = m.group("entity"), m.group("key"), m.group("qualified") + # The key group is deliberately permissive (``.+``) so real composite and + # quoted keys pass through unmangled. That also lets a path separator in, + # which matters here because the resolver below splices ``(key)/qualified`` + # onto a parent URL and consults the registry about the *parent* only. A key + # carrying ``/`` would therefore address one entity while the gate approved + # another. Validate it as the single path component it is meant to be. + validate_record_key("bound-action record key", key) + return entity, key, qualified def _is_unbound_action(entity_set_name: str) -> bool: @@ -99,6 +108,7 @@ def __init__( company_id: str | None = None, # Shared options timeout: int | None = None, + auth: AuthProvider | None = None, ) -> None: if tenant_id is not None: # Programmatic mode — build a synthetic profile @@ -125,10 +135,17 @@ def __init__( self._transport: BCTransport | None = None self._timeout = timeout or self._config.defaults.timeout + self._injected_auth = auth def _ensure_transport(self) -> BCTransport: if self._transport is None: - auth = self._build_auth(self._profile, self._programmatic_secret, self._config) + # An injected provider wins outright: the caller already holds a token, + # so the profile's auth_method is never consulted. That is what lets a + # `browser` profile work server-side, where there is no local browser + # and no loopback listener to bind. + auth = self._injected_auth or self._build_auth( + self._profile, self._programmatic_secret, self._config + ) self._transport = BCTransport(auth, timeout=self._timeout) return self._transport diff --git a/src/bcli/queries/__init__.py b/src/bcli/queries/__init__.py new file mode 100644 index 0000000..e633edb --- /dev/null +++ b/src/bcli/queries/__init__.py @@ -0,0 +1,62 @@ +"""Reusable saved-query engine. + +Catalog loading, parameter validation, ``${{ }}`` expansion, and discovery +(list/search/info) — the whole pipeline behind ``bcli q key=value``, +extracted from the CLI so any consumer (a workflow step, a remote MCP +server) can run a named, parametrised OData query without hand-rolling it +again or importing CLI code to get at it. + +Nothing here performs an HTTP call or talks to a console/terminal — see +:mod:`bcli_cli.commands.query_cmd` for the thin CLI layer built on top of +this package. + +Typical flow:: + + catalog = load_catalog(Path("~/.config/bcli/queries/tech-prod.yaml")) + name = resolve_query_name(catalog, "customer-by-name") # follows aliases + spec = catalog[name] + params = resolve_params(spec.get("params", {}), {"name": "Fabrikam"}) + resolved = expand_query(spec, params) # -> ResolvedQuery + query = resolved.to_query() # -> bcli.odata.Query + response = await client.get(resolved.endpoint, query=query) +""" + +from __future__ import annotations + +from bcli.queries._catalog import ( + RESERVED_QUERY_NAMES, + load_catalog, + load_catalog_from_mapping, + resolve_alias, + resolve_query_name, +) +from bcli.queries._errors import QueryCatalogError, QueryError, QueryParamError +from bcli.queries._expand import ODATA_FIELDS, ResolvedQuery, expand_query +from bcli.queries._params import VALID_PARAM_TYPES, resolve_params, validate_param +from bcli.workflow._query_search import ( + QueryEntry, + filter_entries, + normalize_queries, + search_entries, +) + +__all__ = [ + "ODATA_FIELDS", + "RESERVED_QUERY_NAMES", + "VALID_PARAM_TYPES", + "QueryCatalogError", + "QueryEntry", + "QueryError", + "QueryParamError", + "ResolvedQuery", + "expand_query", + "filter_entries", + "load_catalog", + "load_catalog_from_mapping", + "normalize_queries", + "resolve_alias", + "resolve_params", + "resolve_query_name", + "search_entries", + "validate_param", +] diff --git a/src/bcli/queries/_catalog.py b/src/bcli/queries/_catalog.py new file mode 100644 index 0000000..e5b1c17 --- /dev/null +++ b/src/bcli/queries/_catalog.py @@ -0,0 +1,97 @@ +"""Saved-query catalog loading — from a YAML file or an already-parsed mapping. + +A catalog is the ``queries:`` block of a bundle's YAML file: a mapping of +query name -> spec (see :mod:`bcli_cli.commands.query_cmd` for the full +per-query schema). Loading never talks to Business Central; it only parses +YAML and checks the two structural invariants a bad catalog can violate — +``queries`` must be a mapping, and no query may be named after a ``bcli q`` +sub-verb (``list``, ``search``, ``find``, ``info``, ``run``), since a query +with one of those names would be unreachable except via ``bcli q run +``. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import yaml + +from bcli.queries._errors import QueryCatalogError + +# Reserved names that `bcli q` sub-verb dispatch consumes. A query whose name +# lives here is unreachable except via `bcli q run `, so catalog +# loading hard-errors on a collision — this is a misconfigured bundle, and +# the right place to catch it is at load time, not at dispatch time. +RESERVED_QUERY_NAMES = frozenset({"list", "search", "find", "info", "run"}) + + +def load_catalog(path: Path) -> dict[str, dict[str, Any]]: + """Load and validate a saved-query catalog from a YAML file. + + Returns ``{}`` if ``path`` doesn't exist — an absent catalog is a normal + state (a profile with no saved queries yet), not an error. + + Raises :class:`QueryCatalogError` if the file fails to parse, or if the + parsed structure fails :func:`load_catalog_from_mapping`'s checks. + """ + if not path.is_file(): + return {} + try: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + except yaml.YAMLError as e: + raise QueryCatalogError(f"Failed to parse {path}: {e}") from e + return load_catalog_from_mapping(raw, source=path) + + +def load_catalog_from_mapping( + raw: Mapping[str, Any] | None, + *, + source: str | Path | None = None, +) -> dict[str, dict[str, Any]]: + """Validate an already-parsed catalog mapping (the ``{"queries": {...}}`` shape). + + Use this when the YAML has already been parsed elsewhere (a config + service, a test fixture) and only the structural checks are needed. + ``source`` is used purely to make error messages actionable — pass the + file path (or any label) it came from, when known. + """ + label = str(source) if source is not None else "saved-query catalog" + queries = (raw or {}).get("queries", {}) + if not isinstance(queries, dict): + raise QueryCatalogError(f"{label}: 'queries' must be a mapping.") + + collisions = sorted(set(queries) & RESERVED_QUERY_NAMES) + if collisions: + raise QueryCatalogError( + f"{label}: reserved query names used: {', '.join(collisions)}.\n" + f"These names collide with `bcli q` sub-verbs. Rename the queries " + f"or invoke them via `bcli q run `. " + f"Reserved: {sorted(RESERVED_QUERY_NAMES)}" + ) + + return queries + + +def resolve_alias(catalog: Mapping[str, Mapping[str, Any]], term: str) -> str | None: + """Return the canonical query name when ``term`` matches a declared alias.""" + term_lower = term.lower() + for q_name, body in catalog.items(): + aliases = body.get("aliases") or [] + if not isinstance(aliases, (list, tuple)): + continue + if any(str(a).lower() == term_lower for a in aliases): + return q_name + return None + + +def resolve_query_name(catalog: Mapping[str, Mapping[str, Any]], name: str) -> str | None: + """Return the canonical name for ``name`` — itself, or the query an alias points to. + + ``None`` means neither a direct name nor an alias matched anything in + ``catalog``. + """ + if name in catalog: + return name + return resolve_alias(catalog, name) diff --git a/src/bcli/queries/_errors.py b/src/bcli/queries/_errors.py new file mode 100644 index 0000000..f4a567d --- /dev/null +++ b/src/bcli/queries/_errors.py @@ -0,0 +1,39 @@ +"""Exceptions raised by the saved-query SDK layer. + +These are purely local, pre-HTTP failures — a malformed catalog, a missing +or mistyped parameter — never a response from Business Central. They're +kept separate from the transport-facing errors in :mod:`bcli.errors` (which +map to HTTP status codes) even though they share the same base class, so a +caller can tell "this never left the machine" from "BC rejected it" just by +the exception type. A CLI or an MCP server catches these and decides how to +present them; nothing in this module talks to a console. +""" + +from __future__ import annotations + +from bcli.errors import BCLIError + + +class QueryError(BCLIError): + """Base class for saved-query errors that never reach Business Central.""" + + +class QueryCatalogError(QueryError): + """A saved-query catalog (YAML file or already-parsed mapping) is malformed.""" + + +class QueryParamError(QueryError): + """A supplied parameter is missing, mistyped, or fails its declared constraint. + + ``key`` names the offending parameter. ``kind`` is a coarse, + machine-readable discriminant for callers that want to branch without + parsing the message: ``"schema"`` means the *catalog* is malformed (an + unknown declared type, a non-list enum, …) — a bundle-authoring bug, not + a caller mistake; ``"missing_required"`` and the default ``"value"`` + describe the caller's supplied value. + """ + + def __init__(self, key: str, message: str, *, kind: str = "value") -> None: + self.key = key + self.kind = kind + super().__init__(message) diff --git a/src/bcli/queries/_expand.py b/src/bcli/queries/_expand.py new file mode 100644 index 0000000..2973e63 --- /dev/null +++ b/src/bcli/queries/_expand.py @@ -0,0 +1,85 @@ +"""Resolve a saved-query spec + supplied params into a request spec. + +Nothing here performs an HTTP call: :func:`expand_query` returns a +:class:`ResolvedQuery` — an endpoint plus OData options — that the caller +turns into a request however it likes. ``ResolvedQuery.to_query()`` builds +a :class:`bcli.odata.Query` for callers that already hold a BC client +(the CLI, a workflow step, a remote MCP tool); a caller with different +needs can just read the fields. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from bcli.odata import Query, escape_odata_string +from bcli.workflow import WorkflowContext, resolve_references + +# The seven fields of a saved-query spec that describe the OData request +# itself, plus ``endpoint``. Everything else in a spec (description, +# aliases, tags, owner, …) is discoverability metadata handled by +# :mod:`bcli.workflow._query_search`, not by expansion. +ODATA_FIELDS = ("endpoint", "filter", "select", "expand", "orderby", "top", "skip", "all") + + +@dataclass(frozen=True) +class ResolvedQuery: + """A saved query after ``${{ params.X }}`` substitution — ready to execute.""" + + endpoint: Any = None + filter: Any = None + select: Any = None + expand: Any = None + orderby: Any = None + top: Any = None + skip: Any = None + all: Any = None + + @property + def all_pages(self) -> bool: + """Whether the query should page through every result (the ``all:`` flag).""" + return bool(self.all) + + def to_query(self) -> Query: + """Build a :class:`bcli.odata.Query` from the resolved OData fields.""" + query = Query() + if self.filter: + query.filter(str(self.filter)) + if self.select: + query.select(*[s.strip() for s in str(self.select).split(",")]) + if self.expand: + query.expand(*[e.strip() for e in str(self.expand).split(",")]) + if self.orderby: + query.orderby(str(self.orderby)) + if self.top is not None: + query.top(int(self.top)) + if self.skip is not None: + query.skip(int(self.skip)) + return query + + +def expand_query(spec: Mapping[str, Any], params: Mapping[str, Any]) -> ResolvedQuery: + """Resolve ``${{ params.X }}`` references inside a saved-query spec. + + The ``filter:`` field is re-resolved with OData-escaped string params so + a value containing ``'`` cannot break out of the surrounding string + literal. Other fields (``select``, ``orderby``, ``top``, ``skip``, + ``all``, ``endpoint``) are resolved with raw params — they don't sit + inside OData string literals, so escaping there would corrupt the value + (e.g. an apostrophe in a vendor name passed through for display). + """ + subset = {k: spec[k] for k in ODATA_FIELDS if k in spec} + context = WorkflowContext(params=dict(params)) + expanded = resolve_references(subset, context) + + filter_template = spec.get("filter") + if isinstance(filter_template, str) and "${{" in filter_template: + escaped_params = { + k: (escape_odata_string(v) if isinstance(v, str) else v) for k, v in params.items() + } + filter_ctx = WorkflowContext(params=escaped_params) + expanded["filter"] = resolve_references(filter_template, filter_ctx) + + return ResolvedQuery(**expanded) diff --git a/src/bcli/queries/_params.py b/src/bcli/queries/_params.py new file mode 100644 index 0000000..afab966 --- /dev/null +++ b/src/bcli/queries/_params.py @@ -0,0 +1,169 @@ +"""Saved-query parameter merging + pre-HTTP validation. + +Validation happens entirely client-side, before any request is built. This +is the property that keeps a malformed or hostile value — e.g. an +unescaped ``'`` meant for an OData ``$filter`` — from ever reaching +Business Central; a bad value fails locally with a clear message instead of +round-tripping to BC for a 400. See :mod:`bcli.queries._expand` for the +OData-escaping half of that story (applied once a value has already passed +here). +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping +from typing import Any + +from bcli.queries._errors import QueryParamError + +# Param types a saved-query declaration may declare. +VALID_PARAM_TYPES = frozenset({"string", "integer", "number", "boolean"}) + + +def resolve_params( + declared: Mapping[str, Any] | None, + supplied: Mapping[str, Any] | None, +) -> dict[str, Any]: + """Merge declared defaults with caller-supplied values, then validate. + + ``supplied`` values are already Python-typed (an ``int`` for a limit, a + ``bool`` for a flag, …) — turning raw ``key=value`` CLI strings into + typed values is the caller's job. A non-CLI caller (a workflow step, an + MCP tool invocation) already has typed values and shouldn't have to + round-trip through strings just to reach this function. + + Validation order: + 1. Defaults and supplied values are merged (supplied wins). + 2. Required params are checked. + 3. Each value is coerced/checked against its declared ``type``, + ``pattern``, ``min``, ``max``, ``enum`` constraints. + + Raises :class:`QueryParamError` on the first failure. + """ + declared = declared or {} + resolved: dict[str, Any] = {} + + for key, defn in declared.items(): + if isinstance(defn, dict): + if "default" in defn and defn["default"] is not None: + resolved[key] = defn["default"] + else: + resolved[key] = defn + + resolved.update(supplied or {}) + + for key, defn in declared.items(): + required = (isinstance(defn, dict) and defn.get("required", False)) or ( + not isinstance(defn, dict) and defn is None + ) + if required and key not in resolved: + raise QueryParamError( + key, + f"Missing required parameter '{key}'. " + f"Pass it as: bcli q {key}=", + kind="missing_required", + ) + + for key, defn in declared.items(): + if not isinstance(defn, dict) or key not in resolved: + continue + resolved[key] = validate_param(key, resolved[key], defn) + + return resolved + + +def validate_param(key: str, value: Any, defn: Mapping[str, Any]) -> Any: + """Coerce and validate a single param against its declared constraints. + + When ``type`` is omitted the value is left untouched (preserves + whatever typing the caller already applied, e.g. ``top=5`` staying an + ``int``). When ``type`` is declared, the value is coerced to that type + and the matching constraints (``pattern``, ``min``, ``max``, ``enum``) + are enforced. + + Raises :class:`QueryParamError` naming the param and the rule that + failed. ``kind="schema"`` means the *catalog* declared something + invalid (an unknown type, a non-list enum, a pattern on a non-string + type, a bad regex) — not the caller's input. + """ + type_decl = defn.get("type") + if type_decl is not None and type_decl not in VALID_PARAM_TYPES: + raise QueryParamError( + key, + f"Saved-query schema error: param '{key}' declares unknown type " + f"'{type_decl}'. Valid: {sorted(VALID_PARAM_TYPES)}.", + kind="schema", + ) + + if type_decl == "integer": + try: + value = int(value) + except (TypeError, ValueError): + raise QueryParamError( + key, f"Param '{key}' must be an integer; got {value!r}." + ) from None + elif type_decl == "number": + try: + value = float(value) + except (TypeError, ValueError): + raise QueryParamError( + key, f"Param '{key}' must be a number; got {value!r}." + ) from None + elif type_decl == "boolean": + if isinstance(value, bool): + pass + elif isinstance(value, str) and value.lower() in {"true", "false"}: + value = value.lower() == "true" + else: + raise QueryParamError( + key, f"Param '{key}' must be a boolean (true/false); got {value!r}." + ) + elif type_decl == "string": + value = str(value) + + enum = defn.get("enum") + if enum is not None: + if not isinstance(enum, list): + raise QueryParamError( + key, + f"Saved-query schema error: param '{key}' enum must be a list.", + kind="schema", + ) + if value not in enum: + raise QueryParamError( + key, f"Param '{key}'={value!r} is not in allowed set {enum}." + ) + + pattern = defn.get("pattern") + if pattern is not None: + if type_decl is not None and type_decl != "string": + raise QueryParamError( + key, + f"Saved-query schema error: param '{key}' uses 'pattern' with " + f"type '{type_decl}' (only 'string' supports pattern).", + kind="schema", + ) + try: + matched = re.fullmatch(pattern, str(value)) + except re.error as e: + raise QueryParamError( + key, + f"Saved-query schema error: param '{key}' has invalid regex " + f"{pattern!r}: {e}", + kind="schema", + ) from e + if not matched: + raise QueryParamError( + key, f"Param '{key}'={value!r} does not match pattern {pattern!r}." + ) + + if type_decl in ("integer", "number"): + min_v = defn.get("min") + if min_v is not None and value < min_v: + raise QueryParamError(key, f"Param '{key}'={value} is below min ({min_v}).") + max_v = defn.get("max") + if max_v is not None and value > max_v: + raise QueryParamError(key, f"Param '{key}'={value} exceeds max ({max_v}).") + + return value diff --git a/src/bcli_cli/commands/query_cmd.py b/src/bcli_cli/commands/query_cmd.py index 5060cf4..f44151f 100644 --- a/src/bcli_cli/commands/query_cmd.py +++ b/src/bcli_cli/commands/query_cmd.py @@ -41,21 +41,17 @@ and ``bcli q info`` shows the full record. None of the metadata changes how a query executes. -Defense-in-depth notes: - -* ``type`` / ``pattern`` / ``min`` / ``max`` / ``enum`` are validated *before* - any HTTP call. Bad input fails locally with a clear message instead of - hitting BC and getting back a 400. -* When a string-typed param is interpolated into the ``filter:`` field, OData - single-quote escaping is applied so a value like ``193208' or 1 eq 1--`` - cannot break out of its string literal. The escape is scoped to the filter - context — other fields (``select``, ``top``, etc.) keep raw values. +This module is a thin CLI shell over :mod:`bcli.queries`, which owns the +actual catalog loading, parameter validation, and ``${{ }}`` expansion +logic (so a non-CLI consumer — a workflow step, a remote MCP server — can +reuse it without importing anything under ``bcli_cli``). What stays here is +CLI-specific: parsing ``key=value`` argv into typed values, console +formatting, and exit codes. """ from __future__ import annotations import asyncio -import re from pathlib import Path from typing import Any, Optional @@ -64,7 +60,20 @@ from rich.table import Table from bcli.config._defaults import CONFIG_DIR -from bcli.odata import escape_odata_string +from bcli.queries import ( + ODATA_FIELDS, + QueryCatalogError, + QueryEntry, + QueryParamError, + ResolvedQuery, + expand_query, + filter_entries, + load_catalog, + normalize_queries, + resolve_query_name, + search_entries, +) +from bcli.queries import resolve_params as _sdk_resolve_params from bcli_cli._state import state from bcli_cli.output import format_output, print_context_banner @@ -72,9 +81,6 @@ QUERIES_DIR = CONFIG_DIR / "queries" -# Param types we accept in saved-query declarations. -_VALID_TYPES = frozenset({"string", "integer", "number", "boolean"}) - def query_command( name: Optional[str] = typer.Argument( @@ -118,11 +124,11 @@ def query_command( # sub-verb (`list`, `search`, `find`, `info`), or `run` — the # explicit escape hatch for cases where someone has authored a # query whose name shadows a sub-verb. Reserved names produce a - # hard error at bundle-load time (see `_check_reserved_names` in - # the saved-query loader) so this branch is never reached for a - # well-formed bundle, but `bcli q run ` ensures users always - # have a way to invoke a hypothetically-misnamed query without - # editing the bundle. + # hard error at bundle-load time (see `bcli.queries.load_catalog`, + # which checks `RESERVED_QUERY_NAMES`) so this branch is never + # reached for a well-formed bundle, but `bcli q run ` ensures + # users always have a way to invoke a hypothetically-misnamed query + # without editing the bundle. if name == "run": if not params: console.print("[red]`bcli q run [key=value …]` expected.[/red]") @@ -161,27 +167,22 @@ def query_command( return saved = _load_saved_queries(queries_file) - if name not in saved: - # Try alias resolution before giving up — a curated bundle uses - # aliases to bridge "the query is called overdue-ic but the user - # typed overdue-intercompany" without forcing duplicate definitions. - alias_hit = _resolve_alias(saved, name) - if alias_hit is not None: - name = alias_hit - else: - available = ", ".join(sorted(saved.keys())) or "(none)" - console.print( - f"[red]Saved query '{name}' not found.[/red] Available: {available}\n" - f"[dim]Edit {queries_file} to add one," - f" or run `bcli q search '{name}'` to find a near match.[/dim]" - ) - raise typer.Exit(1) + canonical = resolve_query_name(saved, name) + if canonical is None: + available = ", ".join(sorted(saved.keys())) or "(none)" + console.print( + f"[red]Saved query '{name}' not found.[/red] Available: {available}\n" + f"[dim]Edit {queries_file} to add one," + f" or run `bcli q search '{name}'` to find a near match.[/dim]" + ) + raise typer.Exit(1) + name = canonical spec = saved[name] resolved_params = _resolve_params(spec.get("params", {}), params or []) - resolved = _expand_query(spec, resolved_params) + resolved = expand_query(spec, resolved_params) - endpoint = resolved.get("endpoint") + endpoint = resolved.endpoint if not endpoint: console.print(f"[red]Saved query '{name}' has no 'endpoint'.[/red]") raise typer.Exit(1) @@ -215,13 +216,13 @@ def query_command( capture_filter = state.config.telemetry.capture_filter_text sink.emit(*_tev.query( endpoint=endpoint, - has_filter=bool(resolved.get("filter")), - top=int(resolved.get("top", -1)) if resolved.get("top") not in (None, "") else -1, - skip=int(resolved.get("skip", -1)) if resolved.get("skip") not in (None, "") else -1, - all_pages=bool(resolved.get("all")), + has_filter=bool(resolved.filter), + top=int(resolved.top) if resolved.top not in (None, "") else -1, + skip=int(resolved.skip) if resolved.skip not in (None, "") else -1, + all_pages=resolved.all_pages, status=200, latency_ms=latency_ms, - filter_text=str(resolved.get("filter") or "") if capture_filter else "", + filter_text=str(resolved.filter or "") if capture_filter else "", )) except Exception as e: latency_ms = (_time.monotonic() - started) * 1000.0 @@ -264,8 +265,6 @@ def _list_queries( console.print(f"[dim]{queries_file} has no queries defined.[/dim]") return - from bcli.workflow._query_search import filter_entries, normalize_queries - entries = normalize_queries(queries) entries = filter_entries(entries, tag=tag, owner=owner, freshness=freshness) if not entries: @@ -323,8 +322,6 @@ def _search_queries(profile_name: str, queries_file: Path, phrase: str) -> None: console.print(f"[dim]No saved queries for profile '{profile_name}'.[/dim]") return - from bcli.workflow._query_search import normalize_queries, search_entries - entries = normalize_queries(queries) hits = search_entries(entries, phrase) if not hits: @@ -359,17 +356,14 @@ def _search_queries(profile_name: str, queries_file: Path, phrase: str) -> None: def _query_info(profile_name: str, queries_file: Path, name: str) -> None: """Print full metadata for one query.""" queries = _load_saved_queries(queries_file) - if name not in queries: - alias_hit = _resolve_alias(queries, name) - if alias_hit is None: - console.print( - f"[red]Saved query '{name}' not found.[/red] " - f"Run `bcli q search '{name}'` to find similar." - ) - raise typer.Exit(1) - name = alias_hit - - from bcli.workflow._query_search import QueryEntry + canonical = resolve_query_name(queries, name) + if canonical is None: + console.print( + f"[red]Saved query '{name}' not found.[/red] " + f"Run `bcli q search '{name}'` to find similar." + ) + raise typer.Exit(1) + name = canonical entry = QueryEntry.from_raw(name, queries[name]) @@ -412,18 +406,6 @@ def _query_info(profile_name: str, queries_file: Path, name: str) -> None: console.print(f" [dim]related:[/dim] {', '.join(entry.related)}") -def _resolve_alias(queries: dict[str, dict[str, Any]], term: str) -> str | None: - """Return the canonical query name when ``term`` matches an alias.""" - term_lower = term.lower() - for q_name, body in queries.items(): - aliases = body.get("aliases") or [] - if not isinstance(aliases, (list, tuple)): - continue - if any(str(a).lower() == term_lower for a in aliases): - return q_name - return None - - def _print_starter_example(queries_file: Path) -> None: console.print( "[dim]Example contents (replace with the entities and fields your " @@ -440,261 +422,74 @@ def _print_starter_example(queries_file: Path) -> None: ) -# Reserved names that the `bcli q` sub-verb dispatch consumes. A query -# whose name lives here is unreachable except via `bcli q run `, -# so the loader hard-errors at parse time — this is a misconfigured -# bundle and the right place to catch it is at refresh, not at runtime. -_RESERVED_QUERY_NAMES = frozenset({"list", "search", "find", "info", "run"}) - - def _load_saved_queries(queries_file: Path) -> dict[str, dict[str, Any]]: """Parse a saved-queries YAML file. Returns an empty dict if missing.""" - if not queries_file.is_file(): - return {} - try: - import yaml - except ImportError as e: - console.print("[red]PyYAML is required for saved queries.[/red]") - raise typer.Exit(1) from e - try: - raw = yaml.safe_load(queries_file.read_text(encoding="utf-8")) or {} - except yaml.YAMLError as e: - console.print(f"[red]Failed to parse {queries_file}:[/red] {e}") + return load_catalog(queries_file) + except QueryCatalogError as e: + console.print(f"[red]{e}[/red]") raise typer.Exit(1) from e - queries = raw.get("queries", {}) - if not isinstance(queries, dict): - console.print(f"[red]{queries_file}: 'queries' must be a mapping.[/red]") - raise typer.Exit(1) - - collisions = sorted(set(queries) & _RESERVED_QUERY_NAMES) - if collisions: - console.print( - f"[red]{queries_file}: reserved query names used: " - f"{', '.join(collisions)}.[/red]\n" - f"[dim]These names collide with `bcli q` sub-verbs. " - f"Rename the queries or invoke them via `bcli q run `. " - f"Reserved: {sorted(_RESERVED_QUERY_NAMES)}[/dim]" - ) - raise typer.Exit(1) - - return queries - def _resolve_params( declared: dict[str, Any], cli_args: list[str], ) -> dict[str, Any]: - """Merge declared defaults with ``key=value`` CLI overrides; validate. - - Validation order: - 1. Defaults and CLI args are merged. - 2. Required params are checked. - 3. Each value is coerced/checked against its declared ``type``, - ``pattern``, ``min``, ``max``, ``enum`` constraints. - - Failures exit with a clear, non-developer-friendly error message before any - HTTP call. + """Turn ``key=value`` argv into typed values, then merge + validate. + + The ``key=value`` splitting and type-guessing is CLI-specific (a + non-CLI caller already has typed values); everything past that — + merging with declared defaults, required checks, type/pattern/enum + validation — is :func:`bcli.queries.resolve_params`, which raises + :class:`QueryParamError` instead of exiting directly so it stays usable + outside a CLI context. """ from bcli_cli.commands.batch_cmd import _smart_parse_value - resolved: dict[str, Any] = {} - - for key, defn in (declared or {}).items(): - if isinstance(defn, dict): - if "default" in defn and defn["default"] is not None: - resolved[key] = defn["default"] - else: - resolved[key] = defn - + supplied: dict[str, Any] = {} for arg in cli_args: if "=" not in arg: console.print(f"[red]Invalid parameter '{arg}' — expected key=value.[/red]") raise typer.Exit(1) key, _, raw_value = arg.partition("=") - resolved[key.strip()] = _smart_parse_value(raw_value.strip()) - - for key, defn in (declared or {}).items(): - required = ( - isinstance(defn, dict) and defn.get("required", False) - ) or ( - not isinstance(defn, dict) and defn is None - ) - if required and key not in resolved: - console.print( - f"[red]Missing required parameter '{key}'.[/red] " - f"Pass it as: bcli q {key}=" - ) - raise typer.Exit(1) - - for key, defn in (declared or {}).items(): - if not isinstance(defn, dict) or key not in resolved: - continue - resolved[key] = _validate_param(key, resolved[key], defn) - - return resolved - - -def _validate_param(key: str, value: Any, defn: dict[str, Any]) -> Any: - """Coerce and validate a single param against its declared constraints. - - When ``type`` is omitted the value is left untouched (preserves the smart - parsing applied at CLI parse time, e.g. ``top=5`` stays an ``int``). When - ``type`` is declared, the value is coerced to that type and the matching - constraints (``pattern``, ``min``, ``max``, ``enum``) are enforced. - - Exits via Typer on validation failure with a message that names the param - and the rule that failed. - """ - type_decl = defn.get("type") - if type_decl is not None and type_decl not in _VALID_TYPES: - console.print( - f"[red]Saved-query schema error: param '{key}' declares " - f"unknown type '{type_decl}'. Valid: {sorted(_VALID_TYPES)}.[/red]" - ) - raise typer.Exit(1) + supplied[key.strip()] = _smart_parse_value(raw_value.strip()) - if type_decl == "integer": - try: - value = int(value) - except (TypeError, ValueError): - console.print( - f"[red]Param '{key}' must be an integer; got {value!r}.[/red]" - ) - raise typer.Exit(1) from None - elif type_decl == "number": - try: - value = float(value) - except (TypeError, ValueError): - console.print( - f"[red]Param '{key}' must be a number; got {value!r}.[/red]" - ) - raise typer.Exit(1) from None - elif type_decl == "boolean": - if isinstance(value, bool): - pass - elif isinstance(value, str) and value.lower() in {"true", "false"}: - value = value.lower() == "true" - else: - console.print( - f"[red]Param '{key}' must be a boolean (true/false); " - f"got {value!r}.[/red]" - ) - raise typer.Exit(1) - elif type_decl == "string": - value = str(value) - - enum = defn.get("enum") - if enum is not None: - if not isinstance(enum, list): - console.print( - f"[red]Saved-query schema error: param '{key}' enum must be a list.[/red]" - ) - raise typer.Exit(1) - if value not in enum: - console.print( - f"[red]Param '{key}'={value!r} is not in allowed set {enum}.[/red]" - ) - raise typer.Exit(1) - - pattern = defn.get("pattern") - if pattern is not None: - if type_decl is not None and type_decl != "string": - console.print( - f"[red]Saved-query schema error: param '{key}' uses 'pattern' " - f"with type '{type_decl}' (only 'string' supports pattern).[/red]" - ) - raise typer.Exit(1) - try: - if not re.fullmatch(pattern, str(value)): - console.print( - f"[red]Param '{key}'={value!r} does not match pattern " - f"{pattern!r}.[/red]" - ) - raise typer.Exit(1) - except re.error as e: - console.print( - f"[red]Saved-query schema error: param '{key}' has invalid " - f"regex {pattern!r}: {e}[/red]" - ) - raise typer.Exit(1) from e - - if type_decl in ("integer", "number"): - min_v = defn.get("min") - if min_v is not None and value < min_v: - console.print( - f"[red]Param '{key}'={value} is below min ({min_v}).[/red]" - ) - raise typer.Exit(1) - max_v = defn.get("max") - if max_v is not None and value > max_v: - console.print( - f"[red]Param '{key}'={value} exceeds max ({max_v}).[/red]" - ) - raise typer.Exit(1) - - return value + try: + return _sdk_resolve_params(declared, supplied) + except QueryParamError as e: + console.print(f"[red]{e}[/red]") + raise typer.Exit(1) from e def _expand_query(spec: dict[str, Any], params: dict[str, Any]) -> dict[str, Any]: - """Resolve ``${{ params.X }}`` references inside the saved query spec. - - The ``filter:`` field is re-resolved with OData-escaped string params so a - value containing ``'`` cannot break out of the surrounding string literal. - Other fields (``select``, ``orderby``, ``top``, ``skip``, ``all``, - ``endpoint``) are resolved with raw params — they don't sit inside OData - string literals, so escaping there would corrupt the value (e.g. an - apostrophe in a vendor name passed via ``--show``). - """ - from bcli.workflow._models import WorkflowContext - from bcli.workflow._resolver import resolve_references - - context = WorkflowContext(params=params) - expanded = resolve_references(spec, context) + """Dict-returning wrapper over :func:`bcli.queries.expand_query`. - filter_template = spec.get("filter") - if isinstance(filter_template, str) and "${{" in filter_template: - escaped = { - k: (escape_odata_string(v) if isinstance(v, str) else v) - for k, v in params.items() - } - filter_ctx = WorkflowContext(params=escaped) - expanded["filter"] = resolve_references(filter_template, filter_ctx) - - return expanded + Kept for existing callers/tests that expect the historical + dict-in/dict-out shape; only includes the OData fields the input + ``spec`` actually declared, mirroring what the pre-extraction + implementation returned. + """ + resolved = expand_query(spec, params) + return {key: getattr(resolved, key) for key in ODATA_FIELDS if key in spec} -def _print_resolved(name: str, endpoint: str, resolved: dict[str, Any]) -> None: +def _print_resolved(name: str, endpoint: Any, resolved: ResolvedQuery) -> None: """Show the OData equivalent of a resolved saved query.""" console.print(f"[bold]{name}[/bold] → GET {endpoint}") - for key in ("filter", "select", "expand", "orderby", "top", "skip", "all"): - if key in resolved: - console.print(f" {key}: {resolved[key]}") + for key in ODATA_FIELDS: + if key == "endpoint": + continue + value = getattr(resolved, key) + if value is not None: + console.print(f" {key}: {value}") -async def _run_saved_query(endpoint: str, resolved: dict[str, Any]) -> list[dict]: +async def _run_saved_query(endpoint: str, resolved: ResolvedQuery) -> list[dict]: """Execute the resolved query against the active profile.""" - from bcli.odata._query import Query - - query = Query() - if resolved.get("filter"): - query.filter(str(resolved["filter"])) - if resolved.get("select"): - query.select(*[s.strip() for s in str(resolved["select"]).split(",")]) - if resolved.get("expand"): - query.expand(*[e.strip() for e in str(resolved["expand"]).split(",")]) - if resolved.get("orderby"): - query.orderby(str(resolved["orderby"])) - if resolved.get("top") is not None: - query.top(int(resolved["top"])) - if resolved.get("skip") is not None: - query.skip(int(resolved["skip"])) - - all_pages = bool(resolved.get("all")) + query = resolved.to_query() async with state.make_async_client() as client: - if all_pages: + if resolved.all_pages: records: list[dict] = [] bound = client.query(endpoint) for f in query._params.filters: diff --git a/tests/test_auth/test_static_auth.py b/tests/test_auth/test_static_auth.py new file mode 100644 index 0000000..443f405 --- /dev/null +++ b/tests/test_auth/test_static_auth.py @@ -0,0 +1,97 @@ +"""Tests for StaticTokenAuth — supplying an already-acquired access token. + +This exists for embedders that obtain a BC token themselves and hand it to the +SDK: a hosted service doing an on-behalf-of exchange per request, a CI job with a +token from its platform, a notebook pasting one in. Those callers cannot use the +browser flow, which needs a local browser and a loopback listener on the machine +running the process. + +The token supplier form matters as much as the fixed-string form: a long-lived +process wants to hand over a *callable* so each request picks up a refreshed +token, rather than pinning one that expires. +""" + +from __future__ import annotations + +import pytest + +from bcli.auth._static import StaticTokenAuth +from bcli.errors import ConfigError + + +class TestFixedToken: + async def test_returns_the_token(self): + auth = StaticTokenAuth("header.payload.signature") + assert await auth.get_access_token() == "header.payload.signature" + + async def test_returns_the_same_token_on_repeated_calls(self): + auth = StaticTokenAuth("tok") + assert await auth.get_access_token() == "tok" + assert await auth.get_access_token() == "tok" + + def test_rejects_an_empty_token(self): + """Validate at the boundary — an empty bearer would fail as a confusing 401 + several layers away.""" + with pytest.raises(ConfigError, match="empty"): + StaticTokenAuth("") + + def test_rejects_a_whitespace_only_token(self): + with pytest.raises(ConfigError, match="empty"): + StaticTokenAuth(" ") + + def test_strips_surrounding_whitespace(self): + assert StaticTokenAuth(" tok\n")._token == "tok" + + def test_rejects_a_bearer_prefixed_token(self): + """The transport adds `Bearer ` itself; accepting it here would produce + `Authorization: Bearer Bearer …`.""" + with pytest.raises(ConfigError, match="Bearer"): + StaticTokenAuth("Bearer tok") + + +class TestTokenSupplier: + async def test_calls_an_async_supplier(self): + async def supply() -> str: + return "from-supplier" + + assert await StaticTokenAuth(supply).get_access_token() == "from-supplier" + + async def test_calls_a_sync_supplier(self): + assert await StaticTokenAuth(lambda: "sync-token").get_access_token() == "sync-token" + + async def test_re_invokes_the_supplier_each_call_so_refreshes_are_picked_up(self): + tokens = iter(["first", "second"]) + + async def supply() -> str: + return next(tokens) + + auth = StaticTokenAuth(supply) + assert await auth.get_access_token() == "first" + assert await auth.get_access_token() == "second" + + async def test_rejects_an_empty_token_from_the_supplier(self): + with pytest.raises(ConfigError, match="empty"): + await StaticTokenAuth(lambda: "").get_access_token() + + +class TestProtocolConformance: + async def test_satisfies_the_auth_provider_shape(self): + """Duck-typed against AuthProvider (bcli/auth/_base.py): an awaitable + get_access_token and a synchronous clear_cache.""" + auth = StaticTokenAuth("tok") + assert await auth.get_access_token() == "tok" + assert auth.clear_cache() is None + + async def test_clear_cache_does_not_invalidate_the_supplied_token(self): + """Nothing is cached here — the caller owns the token's lifetime. Callers + that need invalidation should supply a callable instead.""" + auth = StaticTokenAuth("tok") + auth.clear_cache() + assert await auth.get_access_token() == "tok" + + def test_is_usable_as_a_transport_auth_provider(self): + from bcli.client._transport import BCTransport + + auth = StaticTokenAuth("tok") + transport = BCTransport(auth) + assert transport._auth is auth diff --git a/tests/test_client/test_injected_auth.py b/tests/test_client/test_injected_auth.py new file mode 100644 index 0000000..ca348a8 --- /dev/null +++ b/tests/test_client/test_injected_auth.py @@ -0,0 +1,123 @@ +"""Tests for AsyncBCClient(auth=...) — handing the client a ready auth provider. + +Before this, `_ensure_transport` always built an auth provider from the profile's +`auth_method`, so a profile marked `browser` could only ever authenticate by +opening a browser and binding a loopback listener on the host running the process. +That makes the SDK unusable from a server, where the human is somewhere else +entirely. + +Injecting a provider is the seam. The profile still supplies everything else — +environment, company, registry, and the `disable_standard_api` gate — so the +security-relevant routing behaviour is unchanged; only credential acquisition +moves out. +""" + +from __future__ import annotations + +import pytest + +from bcli.auth._static import StaticTokenAuth +from bcli.client._async import AsyncBCClient +from bcli.config._model import BCConfig, BCProfile + + +def _client( + *, + auth=None, + auth_method: str = "browser", + profile_name: str = "test", +) -> AsyncBCClient: + profile = BCProfile( + tenant_id="t1", + environment="Sandbox", + company_id="company-guid-000", + client_id="cid", + auth_method=auth_method, + ) + config = BCConfig(profiles={profile_name: profile}) + config.defaults.profile = profile_name + return AsyncBCClient(profile=profile_name, config=config, auth=auth) + + +class TestInjectedAuthIsUsed: + def test_transport_receives_the_injected_provider(self): + auth = StaticTokenAuth("tok") + transport = _client(auth=auth)._ensure_transport() + assert transport._auth is auth + + def test_injected_auth_overrides_a_browser_profile(self): + """The decisive case: a `browser` profile must not try to open a browser + when the caller already supplied a token.""" + auth = StaticTokenAuth("tok") + transport = _client(auth=auth, auth_method="browser")._ensure_transport() + assert transport._auth is auth + assert type(transport._auth).__name__ != "BrowserAuth" + + def test_injected_auth_overrides_client_credentials_without_needing_a_secret(self): + """No BCLI_CLIENT_SECRET, no keyring entry — and no error, because the + credential path is bypassed entirely.""" + auth = StaticTokenAuth("tok") + transport = _client(auth=auth, auth_method="client_credentials")._ensure_transport() + assert transport._auth is auth + + def test_injected_auth_bypasses_auth_method_validation(self): + """An unsupported `auth_method` normally raises ConfigError. With injected + auth it is never consulted, so a profile that only ever runs server-side + need not name a flow it cannot perform.""" + transport = _client(auth=StaticTokenAuth("tok"), auth_method="nonsense")._ensure_transport() + assert isinstance(transport._auth, StaticTokenAuth) + + def test_transport_is_built_once_and_reused(self): + client = _client(auth=StaticTokenAuth("tok")) + assert client._ensure_transport() is client._ensure_transport() + + +class TestExistingBehaviourPreserved: + def test_browser_profile_still_builds_browser_auth_when_nothing_injected(self): + transport = _client(auth_method="browser")._ensure_transport() + assert type(transport._auth).__name__ == "BrowserAuth" + + def test_device_code_profile_still_builds_device_code_auth(self): + transport = _client(auth_method="device_code")._ensure_transport() + assert type(transport._auth).__name__ == "DeviceCodeAuth" + + def test_unsupported_auth_method_still_raises_without_injection(self): + from bcli.errors import ConfigError + + with pytest.raises(ConfigError, match="Unsupported auth_method"): + _client(auth_method="nonsense")._ensure_transport() + + +class TestProgrammaticMode: + def test_injected_auth_works_without_any_config_file(self): + auth = StaticTokenAuth("tok") + client = AsyncBCClient( + tenant_id="t1", + environment="Production", + company_id="company-guid-000", + auth=auth, + ) + assert client._ensure_transport()._auth is auth + + +class TestRegistryGateStillApplies: + def test_disable_standard_api_still_blocks_unknown_entities(self): + """Injecting auth must not weaken the routing gate — that lives on the + profile and registry, not on the credential path.""" + from bcli.errors import RegistryError + + profile = BCProfile( + tenant_id="t1", + environment="Sandbox", + company_id="company-guid-000", + client_id="cid", + disable_standard_api=True, + ) + config = BCConfig(profiles={"p": profile}) + config.defaults.profile = "p" + client = AsyncBCClient(profile="p", config=config, auth=StaticTokenAuth("tok")) + + with pytest.raises(RegistryError): + client._resolve_url_for_target( + "Sandbox", "company-guid-000", "definitelyNotInAnyRegistry" + ) diff --git a/tests/test_context/test_bundle.py b/tests/test_context/test_bundle.py index e3624bf..3160c86 100644 --- a/tests/test_context/test_bundle.py +++ b/tests/test_context/test_bundle.py @@ -133,9 +133,15 @@ def test_audit_trail_complete_for_layered_redactions() -> None: assert "context:url_query" in rule_ids -def test_no_context_policy_path() -> None: +def test_no_context_policy_path(tmp_path: Path) -> None: # Mimic `bcli ask --no-context`: caller suppresses describe + tail # via policy and provides nothing else. + # + # `--no-context` also passes skip_last_error=True (ask_cmd.py), which is what + # actually suppresses the implicit read of last-error.json. Without it — and + # without isolating config_dir the way the other tests here do — this asserted + # against the developer's real ~/.config/bcli and failed on any machine that + # had ever recorded a bcli error. bundle = build_bundle( question="just answer", policy=BundlePolicy( @@ -143,6 +149,8 @@ def test_no_context_policy_path() -> None: include_http_tail=False, include_bodies=False, ), + config_dir=tmp_path, + skip_last_error=True, ) kinds = {s.kind for s in bundle.sources} assert kinds == {"question"} diff --git a/tests/test_queries/__init__.py b/tests/test_queries/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_queries/test_catalog.py b/tests/test_queries/test_catalog.py new file mode 100644 index 0000000..c531955 --- /dev/null +++ b/tests/test_queries/test_catalog.py @@ -0,0 +1,127 @@ +"""Tests for bcli.queries catalog loading (YAML file + already-parsed mapping).""" + +from __future__ import annotations + +import textwrap + +import pytest + +from bcli.queries import ( + RESERVED_QUERY_NAMES, + QueryCatalogError, + load_catalog, + load_catalog_from_mapping, + resolve_alias, + resolve_query_name, +) + +# ── load_catalog (YAML file) ──────────────────────────────────────────── + + +def test_load_catalog_missing_file_returns_empty(tmp_path): + assert load_catalog(tmp_path / "nope.yaml") == {} + + +def test_load_catalog_parses_valid_file(tmp_path): + f = tmp_path / "team.yaml" + f.write_text( + textwrap.dedent("""\ + queries: + customer-by-name: + description: Look up a customer by display name + endpoint: customers + params: + name: + required: true + filter: "displayName eq '${{ params.name }}'" + orderby: displayName asc + top: 25 + """) + ) + queries = load_catalog(f) + assert "customer-by-name" in queries + spec = queries["customer-by-name"] + assert spec["endpoint"] == "customers" + assert spec["top"] == 25 + + +def test_load_catalog_empty_file_returns_empty(tmp_path): + f = tmp_path / "empty.yaml" + f.write_text("") + assert load_catalog(f) == {} + + +def test_load_catalog_rejects_malformed_yaml(tmp_path): + f = tmp_path / "bad.yaml" + f.write_text("queries: [unterminated\n") + with pytest.raises(QueryCatalogError, match="Failed to parse"): + load_catalog(f) + + +def test_load_catalog_rejects_non_mapping_queries(tmp_path): + f = tmp_path / "bad.yaml" + f.write_text("queries:\n - just-a-list-item\n") + with pytest.raises(QueryCatalogError, match="'queries' must be a mapping"): + load_catalog(f) + + +def test_load_catalog_rejects_reserved_names(tmp_path): + f = tmp_path / "bad.yaml" + f.write_text("queries:\n list:\n endpoint: customers\n") + with pytest.raises(QueryCatalogError, match="reserved query names"): + load_catalog(f) + + +# ── load_catalog_from_mapping ─────────────────────────────────────────── + + +def test_load_catalog_from_mapping_accepts_already_parsed_dict(): + raw = {"queries": {"foo": {"endpoint": "vendors"}}} + assert load_catalog_from_mapping(raw) == {"foo": {"endpoint": "vendors"}} + + +def test_load_catalog_from_mapping_none_returns_empty(): + assert load_catalog_from_mapping(None) == {} + + +def test_load_catalog_from_mapping_error_includes_source_label(): + with pytest.raises(QueryCatalogError, match="my-source"): + load_catalog_from_mapping({"queries": ["nope"]}, source="my-source") + + +def test_reserved_query_names_constant(): + assert {"list", "search", "find", "info", "run"} == RESERVED_QUERY_NAMES + + +# ── resolve_alias / resolve_query_name ────────────────────────────────── + + +_CATALOG = { + "overdue-ic": {"aliases": ["overdue-intercompany", "IC-Overdue"]}, + "open-pos": {}, +} + + +def test_resolve_alias_matches_case_insensitively(): + assert resolve_alias(_CATALOG, "ic-overdue") == "overdue-ic" + + +def test_resolve_alias_no_match_returns_none(): + assert resolve_alias(_CATALOG, "nope") is None + + +def test_resolve_alias_ignores_non_list_aliases(): + catalog = {"foo": {"aliases": "not-a-list"}} + assert resolve_alias(catalog, "not-a-list") is None + + +def test_resolve_query_name_direct_hit(): + assert resolve_query_name(_CATALOG, "open-pos") == "open-pos" + + +def test_resolve_query_name_via_alias(): + assert resolve_query_name(_CATALOG, "overdue-intercompany") == "overdue-ic" + + +def test_resolve_query_name_unknown_returns_none(): + assert resolve_query_name(_CATALOG, "unknown-thing") is None diff --git a/tests/test_queries/test_expand.py b/tests/test_queries/test_expand.py new file mode 100644 index 0000000..f330575 --- /dev/null +++ b/tests/test_queries/test_expand.py @@ -0,0 +1,145 @@ +"""Tests for bcli.queries.expand_query and ResolvedQuery.""" + +from __future__ import annotations + +from bcli.odata import Query +from bcli.queries import ResolvedQuery, expand_query + +# ── expand_query — ${{ params.X }} resolution ──────────────────────────── + + +def test_expand_query_resolves_param_references(): + spec = { + "endpoint": "engineUtilizations", + "filter": "engineSerialNumber eq '${{ params.esn }}'", + "top": 24, + } + resolved = expand_query(spec, {"esn": "193208"}) + assert resolved.filter == "engineSerialNumber eq '193208'" + assert resolved.endpoint == "engineUtilizations" + assert resolved.top == 24 + + +def test_expand_query_preserves_full_reference_type(): + spec = {"endpoint": "x", "top": "${{ params.limit }}"} + resolved = expand_query(spec, {"limit": 50}) + assert resolved.top == 50 + assert isinstance(resolved.top, int) + + +def test_expand_query_absent_fields_stay_none(): + resolved = expand_query({"endpoint": "x"}, {}) + assert resolved.filter is None + assert resolved.select is None + assert resolved.all is None + + +def test_expand_query_ignores_non_odata_metadata_fields(): + """description/aliases/tags/etc. aren't part of the request spec.""" + spec = { + "endpoint": "x", + "description": "irrelevant to expansion", + "tags": ["a", "b"], + "params": {"esn": {"required": True}}, + } + resolved = expand_query(spec, {"esn": "1"}) + assert not hasattr(resolved, "description") + assert not hasattr(resolved, "tags") + + +# ── Filter-context OData escaping ──────────────────────────────────────── + + +def test_expand_query_escapes_single_quote_in_filter(): + spec = {"endpoint": "vendors", "filter": "name eq '${{ params.name }}'"} + resolved = expand_query(spec, {"name": "O'Brien"}) + assert resolved.filter == "name eq 'O''Brien'" + + +def test_expand_query_neutralises_injection_in_filter(): + spec = { + "endpoint": "engineUtilizations", + "filter": "engineSerialNumber eq '${{ params.esn }}'", + } + resolved = expand_query(spec, {"esn": "193208' or 1 eq 1--"}) + assert resolved.filter == "engineSerialNumber eq '193208'' or 1 eq 1--'" + assert resolved.filter.count("'") % 2 == 0 + + +def test_expand_query_does_not_escape_outside_filter(): + spec = { + "endpoint": "items", + "filter": "name eq '${{ params.name }}'", + "select": "${{ params.name }}", + "orderby": "${{ params.name }} asc", + } + resolved = expand_query(spec, {"name": "O'Brien"}) + assert resolved.filter == "name eq 'O''Brien'" + assert resolved.select == "O'Brien" + assert resolved.orderby == "O'Brien asc" + + +def test_expand_query_leaves_filter_alone_when_no_substitution(): + spec = {"endpoint": "x", "filter": "blocked eq false"} + resolved = expand_query(spec, {}) + assert resolved.filter == "blocked eq false" + + +def test_expand_query_non_string_param_in_filter_passes_through(): + spec = {"endpoint": "x", "filter": "amount gt ${{ params.threshold }}"} + resolved = expand_query(spec, {"threshold": 1000}) + assert resolved.filter == "amount gt 1000" + + +# ── ResolvedQuery.to_query() / all_pages ───────────────────────────────── + + +def test_to_query_builds_expected_odata_params(): + resolved = ResolvedQuery( + endpoint="vendors", + filter="name eq 'Acme'", + select="no,name", + expand="ledgerEntries", + orderby="name asc", + top=10, + skip=5, + ) + query = resolved.to_query() + assert isinstance(query, Query) + params = query.to_params() + assert params["$filter"] == "(name eq 'Acme')" + assert params["$select"] == "no,name" + assert params["$expand"] == "ledgerEntries" + assert params["$orderby"] == "name asc" + assert params["$top"] == "10" + assert params["$skip"] == "5" + + +def test_to_query_splits_comma_lists_and_strips_whitespace(): + resolved = ResolvedQuery(endpoint="x", select=" a , b ,c", expand=" nav1 , nav2") + query = resolved.to_query() + assert query._params.selects == ["a", "b", "c"] + assert query._params.expands == ["nav1", "nav2"] + + +def test_to_query_empty_when_no_fields_set(): + resolved = ResolvedQuery(endpoint="x") + assert resolved.to_query().is_empty + + +def test_to_query_top_zero_is_applied_not_skipped(): + """`is not None` semantics — top=0 is a real value, not "unset".""" + resolved = ResolvedQuery(endpoint="x", top=0) + assert resolved.to_query().to_params()["$top"] == "0" + + +def test_all_pages_defaults_false(): + assert ResolvedQuery(endpoint="x").all_pages is False + + +def test_all_pages_true_when_all_set(): + assert ResolvedQuery(endpoint="x", all=True).all_pages is True + + +def test_all_pages_coerces_truthy_values(): + assert ResolvedQuery(endpoint="x", all="yes").all_pages is True diff --git a/tests/test_queries/test_params.py b/tests/test_queries/test_params.py new file mode 100644 index 0000000..ef5979b --- /dev/null +++ b/tests/test_queries/test_params.py @@ -0,0 +1,168 @@ +"""Tests for bcli.queries parameter merging + pre-HTTP validation. + +Mirrors the coverage that lived in tests/test_cli/test_query_cmd.py before +the extraction, at the SDK boundary: `supplied` is already a typed mapping +(no `key=value` string parsing here — that's the CLI's job). +""" + +from __future__ import annotations + +import pytest + +from bcli.queries import QueryParamError, resolve_params, validate_param + +# ── resolve_params ─────────────────────────────────────────────────────── + + +def test_resolve_params_uses_default(): + declared = {"top": {"required": False, "default": 10}} + assert resolve_params(declared, {}) == {"top": 10} + + +def test_resolve_params_supplied_overrides_default(): + declared = {"top": {"required": False, "default": 10}} + assert resolve_params(declared, {"top": 5}) == {"top": 5} + + +def test_resolve_params_required_missing_raises(): + declared = {"esn": {"required": True}} + with pytest.raises(QueryParamError) as exc_info: + resolve_params(declared, {}) + assert exc_info.value.key == "esn" + assert exc_info.value.kind == "missing_required" + assert "Missing required parameter 'esn'" in str(exc_info.value) + + +def test_resolve_params_required_supplied(): + declared = {"esn": {"required": True}} + assert resolve_params(declared, {"esn": 193208}) == {"esn": 193208} + + +def test_resolve_params_none_declared_passes_supplied_through_unvalidated(): + """No schema to check against — supplied values are merged as-is.""" + assert resolve_params(None, {"anything": 1}) == {"anything": 1} + + +def test_resolve_params_none_declared_and_none_supplied_is_empty(): + assert resolve_params(None, None) == {} + + +def test_resolve_params_unknown_supplied_keys_pass_through(): + """Params not in `declared` aren't validated but do end up in the result.""" + declared = {"esn": {"required": True}} + resolved = resolve_params(declared, {"esn": "1", "extra": "kept"}) + assert resolved == {"esn": "1", "extra": "kept"} + + +# ── validate_param via resolve_params (type/pattern/min/max/enum) ──────── + + +class TestParamValidation: + def test_integer_type_coerces_string(self): + declared = {"limit": {"required": True, "type": "integer"}} + resolved = resolve_params(declared, {"limit": "50"}) + assert resolved == {"limit": 50} + assert isinstance(resolved["limit"], int) + + def test_integer_type_rejects_non_integer(self): + declared = {"limit": {"required": True, "type": "integer"}} + with pytest.raises(QueryParamError): + resolve_params(declared, {"limit": "abc"}) + + def test_integer_max_bound_enforced(self): + declared = {"limit": {"required": True, "type": "integer", "max": 1000}} + with pytest.raises(QueryParamError, match="exceeds max"): + resolve_params(declared, {"limit": 99999}) + + def test_integer_min_bound_enforced(self): + declared = {"limit": {"required": True, "type": "integer", "min": 1}} + with pytest.raises(QueryParamError, match="below min"): + resolve_params(declared, {"limit": 0}) + + def test_integer_within_bounds_accepted(self): + declared = {"limit": {"required": True, "type": "integer", "min": 1, "max": 100}} + assert resolve_params(declared, {"limit": 50}) == {"limit": 50} + + def test_string_pattern_accepts_match(self): + declared = {"airline": {"required": True, "type": "string", "pattern": r"^[A-Z0-9]{2,8}$"}} + assert resolve_params(declared, {"airline": "AIRNORTH"}) == {"airline": "AIRNORTH"} + + def test_string_pattern_rejects_non_match(self): + declared = {"airline": {"required": True, "type": "string", "pattern": r"^[A-Z0-9]{2,8}$"}} + with pytest.raises(QueryParamError): + resolve_params(declared, {"airline": "little caesars"}) + + def test_string_pattern_rejects_injection_attempt(self): + declared = {"esn": {"required": True, "type": "string", "pattern": r"^\d{4,8}$"}} + with pytest.raises(QueryParamError): + resolve_params(declared, {"esn": "193208' or 1 eq 1--"}) + + def test_enum_accepts_valid(self): + declared = {"status": {"required": True, "enum": ["Open", "Posted"]}} + assert resolve_params(declared, {"status": "Open"}) == {"status": "Open"} + + def test_enum_rejects_invalid(self): + declared = {"status": {"required": True, "enum": ["Open", "Posted"]}} + with pytest.raises(QueryParamError, match="not in allowed set"): + resolve_params(declared, {"status": "Cancelled"}) + + def test_enum_declared_as_non_list_is_a_schema_error(self): + declared = {"status": {"required": True, "enum": "Open"}} + with pytest.raises(QueryParamError) as exc_info: + resolve_params(declared, {"status": "Open"}) + assert exc_info.value.kind == "schema" + + def test_unknown_type_is_a_schema_error(self): + declared = {"x": {"required": True, "type": "uuid"}} + with pytest.raises(QueryParamError) as exc_info: + resolve_params(declared, {"x": "anything"}) + assert exc_info.value.kind == "schema" + + def test_pattern_on_non_string_type_is_a_schema_error(self): + declared = {"x": {"required": True, "type": "integer", "pattern": r"^\d+$"}} + with pytest.raises(QueryParamError) as exc_info: + resolve_params(declared, {"x": 5}) + assert exc_info.value.kind == "schema" + + def test_invalid_regex_is_a_schema_error(self): + declared = {"x": {"required": True, "type": "string", "pattern": "([unterminated"}} + with pytest.raises(QueryParamError) as exc_info: + resolve_params(declared, {"x": "anything"}) + assert exc_info.value.kind == "schema" + + def test_no_type_defaults_to_string(self): + declared = {"name": {"required": True}} + assert resolve_params(declared, {"name": "Fabrikam"}) == {"name": "Fabrikam"} + + def test_boolean_type_accepts_string_true(self): + declared = {"all": {"required": True, "type": "boolean"}} + assert resolve_params(declared, {"all": "true"}) == {"all": True} + + def test_boolean_type_accepts_native_bool(self): + declared = {"all": {"required": True, "type": "boolean"}} + assert resolve_params(declared, {"all": False}) == {"all": False} + + def test_boolean_type_rejects_other_values(self): + declared = {"all": {"required": True, "type": "boolean"}} + with pytest.raises(QueryParamError): + resolve_params(declared, {"all": "yes"}) + + def test_number_type_coerces(self): + declared = {"rate": {"required": True, "type": "number"}} + assert resolve_params(declared, {"rate": "3.14"}) == {"rate": 3.14} + + def test_number_type_rejects_non_numeric(self): + declared = {"rate": {"required": True, "type": "number"}} + with pytest.raises(QueryParamError): + resolve_params(declared, {"rate": "abc"}) + + +# ── validate_param directly ────────────────────────────────────────────── + + +def test_validate_param_passthrough_when_untyped(): + assert validate_param("k", "raw", {}) == "raw" + + +def test_validate_param_string_coerces_non_string_values(): + assert validate_param("k", 42, {"type": "string"}) == "42" diff --git a/tests/test_url/test_record_key_validation.py b/tests/test_url/test_record_key_validation.py new file mode 100644 index 0000000..b3f79f7 --- /dev/null +++ b/tests/test_url/test_record_key_validation.py @@ -0,0 +1,170 @@ +"""Record keys and entity-set names must not carry raw URL path syntax. + +``build_url`` validated the custom-API route segments (publisher/group/version) +but spliced ``entity_set_name`` and ``record_id`` in raw. A record key is +attacker-influenced in any layer that accepts one from a caller, and a raw ``/`` +in it starts a new path segment — so the composed URL could address an entity +other than the one the registry was consulted about: + + entity_set_name="engineOverviews" + record_id="1)/../../../../../../glEntries('X'" + → .../engineOverviews(1)/../../../../../../glEntries('X') + +which collapses to ``glEntries('X')``. The registry only ever saw +``engineOverviews``, so its decision was made about the wrong target. Business +Central still applies the caller's own permission set, so this is not a way to +read data the user could not otherwise reach — but it does defeat the +client-side registry restriction, and ``bcli/client/_async.py``'s bound-action +resolver documents an assumption ("everything from the ``(`` onward is opaque to +the registry ... gated on the parent, which is the security-relevant identity") +that only holds once keys cannot contain path syntax. + +The fix rejects raw path and query delimiters. Everything else a real OData key +needs — quotes, commas, equals signs, hyphens, parentheses inside a quoted +string — keeps working, because those cannot start a new path segment. +""" + +from __future__ import annotations + +import pytest + +from bcli._url import build_url + +ENV = "SBEnvJun26" +COMPANY = "f99bd320-b400-4189-b3c1-c62c05d4e7a5" + + +def _build(**kw) -> str: + return build_url(environment=ENV, company_id=COMPANY, **kw) + + +class TestRecordIdRejectsPathSyntax: + def test_rejects_the_reported_traversal_payload(self): + with pytest.raises(ValueError, match="record_id"): + _build( + entity_set_name="engineOverviews", + record_id="1)/../../../../../../glEntries('X'", + ) + + @pytest.mark.parametrize( + "bad", + [ + "1/2", + "1)/glEntries('X'", + "..", + "../customers", + "1\\2", + "1)?$filter=true", + "1)#frag", + ], + ) + def test_rejects_path_and_query_delimiters(self, bad: str): + with pytest.raises(ValueError, match="record_id"): + _build(entity_set_name="engineOverviews", record_id=bad) + + def test_error_names_percent_encoding_as_the_remedy(self): + """A caller with a genuine '/' in a key needs to know what to do.""" + with pytest.raises(ValueError, match="percent-encode"): + _build(entity_set_name="engineOverviews", record_id="AB/CD") + + +class TestRecordIdStillAcceptsRealKeys: + @pytest.mark.parametrize( + "good", + [ + "f99bd320-b400-4189-b3c1-c62c05d4e7a5", # GUID + "42", # integer key + "'V00010'", # quoted string + "'O''Brien'", # quoted string, escaped quote + "'ACME (US)'", # parens inside a quoted string are legitimate + "k1='a',k2='b'", # composite key + "%2F", # already percent-encoded separator + ], + ) + def test_accepts(self, good: str): + url = _build(entity_set_name="engineOverviews", record_id=good) + assert url.endswith(f"engineOverviews({good})") + + def test_none_record_id_is_unchanged(self): + url = _build(entity_set_name="engineOverviews") + assert url.endswith("engineOverviews") + + +class TestEntitySetNameIsValidatedToo: + @pytest.mark.parametrize("bad", ["a/b", "..", ".", "a\\b"]) + def test_rejects_path_syntax(self, bad: str): + with pytest.raises(ValueError, match="entity_set_name"): + _build(entity_set_name=bad) + + def test_accepts_a_normal_entity_set(self): + assert _build(entity_set_name="engineOverviews").endswith("engineOverviews") + + +class TestCustomApiRouteStillValidated: + """Pre-existing behaviour must not regress.""" + + def test_publisher_traversal_still_rejected(self): + with pytest.raises(ValueError, match="publisher"): + _build( + entity_set_name="engineOverviews", + publisher="../..", + group="technical", + version="v1.5", + ) + + def test_custom_route_composes(self): + url = _build( + entity_set_name="engineOverviews", + publisher="beautech", + group="technical", + version="v1.5", + ) + assert "/api/beautech/technical/v1.5/" in url + + +class TestBoundActionKeyIsValidated: + """The bound-action resolver splices ``(key)/Namespace.Action`` onto a parent + URL it resolved from the registry. If the key can hold a path separator, the + parent-only registry check is not the boundary the code says it is.""" + + def _client(self): + from bcli.client._async import AsyncBCClient + from bcli.config._model import BCConfig, BCProfile + + profile = BCProfile( + tenant_id="t", + environment=ENV, + company_id=COMPANY, + client_id="c", + disable_standard_api=True, + ) + config = BCConfig(profiles={"p": profile}) + config.defaults.profile = "p" + return AsyncBCClient(profile="p", config=config) + + def test_rejects_a_traversing_key_in_a_bound_action(self): + client = self._client() + with pytest.raises(ValueError, match="record key|record_id"): + client._resolve_url_for_target( + ENV, + COMPANY, + "engineOverviews(1)/../../glEntries('X')/Microsoft.NAV.doThing", + publisher="beautech", + group="technical", + version="v1.5", + ) + + def test_still_resolves_a_legitimate_bound_action(self): + client = self._client() + url = client._resolve_url_for_target( + ENV, + COMPANY, + "engineOverviews(f99bd320-b400-4189-b3c1-c62c05d4e7a5)/Microsoft.NAV.updateLlpUtilization", + publisher="beautech", + group="technical", + version="v1.5", + ) + assert url.endswith( + "engineOverviews(f99bd320-b400-4189-b3c1-c62c05d4e7a5)" + "/Microsoft.NAV.updateLlpUtilization" + ) diff --git a/uv.lock b/uv.lock index 4f20ff6..e13a74a 100644 --- a/uv.lock +++ b/uv.lock @@ -321,7 +321,7 @@ wheels = [ [[package]] name = "bc-cli" -version = "0.6.2" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "httpx" },