From 15aa628742f905929e910270defecbbd4e5f85d7 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 11:44:33 -0400 Subject: [PATCH 1/7] OpenConceptLab/ocl_online#553 | 0.2.1: encode identifiers as path segments, tighten the origin check - Every ID, version, key or queue name the client puts in a request path is quoted as one segment. An empty value, ".", ".." or a value containing "/" raises APIError (status_code 0) before anything is sent. search_repos now validates owner_type, like the other methods. - `ocl tool recreate-collection` quotes the collection paths it builds, and refuses an export whose collection ID would name another endpoint. - The origin check compares hosts in their encoded form, refuses URLs that carry credentials, and checks a base_url path prefix against the path as a server that decodes and normalizes it would route it. A host that isn't a valid domain name raises APIError instead of idna's error. - Version 0.2.1. The README says what the library refuses. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- README.md | 2 + pyproject.toml | 2 +- src/ocl_cli/commands/tool.py | 15 ++- src/openconceptlab/client.py | 165 ++++++++++++++++++------------ tests/test_client.py | 189 +++++++++++++++++++++++++++++++++++ 5 files changed, 308 insertions(+), 65 deletions(-) diff --git a/README.md b/README.md index 07ce642..6ae2e91 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,8 @@ Pass `auth_scheme="Bearer"` to use an OIDC access token instead of an API token. HTTP errors raise `openconceptlab.APIError`, with `status_code`, `detail` and the parsed response `body`. Network failures raise the httpx exception (`httpx.TimeoutException` or `httpx.TransportError`) once retries run out. A client holds one token, so create one per user. +Each ID, version or key you pass is encoded as one segment of the request path. An empty value, `.`, `..` or a value containing `/` raises `APIError` (`status_code` 0) before anything is sent. So does a request to a URL outside the client's `base_url`, such as one taken from a response. + ## Quick Start ### 1. Authenticate diff --git a/pyproject.toml b/pyproject.toml index ed6a330..f1a1ddc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "openconceptlab" -version = "0.2.0" +version = "0.2.1" description = "Python client and command-line interface for the Open Concept Lab API" readme = "README.md" license = "MPL-2.0" diff --git a/src/ocl_cli/commands/tool.py b/src/ocl_cli/commands/tool.py index 2b854d0..d6c410b 100644 --- a/src/ocl_cli/commands/tool.py +++ b/src/ocl_cli/commands/tool.py @@ -14,6 +14,7 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error +from openconceptlab.client import _segment # Regex to detect versioned concept/mapping expressions (e.g. /concepts/100/1/) VERSIONED_EXPR = re.compile(r"/(concepts|mappings)/\d+/\d+/?$") @@ -117,6 +118,12 @@ def _collection_url(org: str | None, collection_id: str) -> str: return f"/user/collections/{collection_id}/" +def _collection_path(org: str | None, collection_id: str) -> str: + """The collection's API path, each identifier encoded as one segment.""" + owner = f"/orgs/{_segment(org)}" if org else "/user" + return f"{owner}/collections/{_segment(collection_id)}/" + + def _build_references(export, org, collection_id, client, callback=None): """Build deduplicated, versionless reference lines from export data. @@ -238,6 +245,11 @@ def recreate_collection(ctx, input_path, org, name, output_path, dry_run): if not collection_id: click.echo("Error: Collection id not found in export and --name not provided.", err=True) sys.exit(1) + try: + collection_path = _collection_path(org, collection_id) + except APIError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) # Default output filename if not output_path: @@ -273,7 +285,6 @@ def recreate_collection(ctx, input_path, org, name, output_path, dry_run): # Delete and recreate collection click.echo("Recreating collection and importing references...") - collection_path = _collection_url(org, collection_id) try: # Check if collection exists try: @@ -289,7 +300,7 @@ def recreate_collection(ctx, input_path, org, name, output_path, dry_run): # Create collection click.echo(f"Creating collection {collection_id}...") payload = _build_collection_payload(export_data, name) - collections_path = f"/orgs/{org}/collections/" if org else "/user/collections/" + collections_path = f"/orgs/{_segment(org)}/collections/" if org else "/user/collections/" client.post(collections_path, json=payload) # Bulk import diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index b427af7..3f511ca 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -6,6 +6,7 @@ import platform import sys from typing import Any, Optional +from urllib.parse import quote import httpx from tenacity import ( @@ -51,6 +52,30 @@ def _repo_type_stem(repo_type: str) -> str: raise ValueError(f"repo_type must be 'source' or 'collection', got '{repo_type}'") +def _segment(value: Any) -> str: + """Encode a caller's value (an ID, a version, a key) as one path segment. + + Refuses a value that would name a different endpoint: empty, ``.``, ``..`` or + containing ``/``, which the server decodes back into a separator. + """ + text = str(value) + if text in ("", ".", "..") or "/" in text: + raise APIError(f"Invalid identifier {text!r}: it can't be used in an API path", status_code=0) + return quote(text, safe="") + + +def _resolve_dot_segments(path: str) -> str: + """The path as a server that decodes and normalizes it would route it.""" + parts: list[str] = [] + for part in path.replace("\\", "/").split("/"): + if part == "..": + if parts: + parts.pop() + elif part not in ("", "."): + parts.append(part) + return "/" + "/".join(parts) + + def _validate_owner_type(owner_type: str) -> None: if owner_type not in ("users", "orgs"): raise ValueError(f"owner_type must be 'users' or 'orgs', got '{owner_type}'") @@ -67,12 +92,20 @@ def _build_repo_endpoint( """Build a repository-scoped endpoint URL.""" _validate_owner_type(owner_type) stem = _repo_type_stem(repo_type) - base = f"/{owner_type}/{owner}/{stem}/{repo}" + base = f"/{owner_type}/{_segment(owner)}/{stem}/{_segment(repo)}" if repo_version: - base = f"{base}/{repo_version}" + base = f"{base}/{_segment(repo_version)}" return f"{base}/{suffix}" if suffix else f"{base}/" +class _HTTPClient(httpx.Client): + def build_request(self, *args: Any, **kwargs: Any) -> httpx.Request: + try: + return super().build_request(*args, **kwargs) + except UnicodeError as e: # idna's error for a host that isn't a valid domain name + raise APIError(f"Refusing to send a request to an invalid host: {e}", status_code=0) from e + + class OCLAPIClient: """Synchronous client for the OCL API.""" @@ -113,7 +146,7 @@ def __init__( merged = httpx.Headers(default_headers) merged.update(headers or {}) - self.client = httpx.Client( + self.client = _HTTPClient( base_url=self.base_url, headers=merged, timeout=timeout, @@ -122,15 +155,23 @@ def __init__( ) def _check_destination(self, request: httpx.Request) -> None: - """Refuse any request outside the API's origin and path: the token goes only there.""" + """Refuse any request outside the API's origin and path: the token goes only there. + + Hosts are compared in their encoded (punycode) form, and the path as a server + that decodes and normalizes it would route it. + """ base, url = self._base, request.url + if url.userinfo: + raise APIError("Refusing to send a request with credentials in its URL", status_code=0) prefix = base.path.rstrip("/") + path = _resolve_dot_segments(url.path) if ( - (url.scheme, url.host, url.port) != (base.scheme, base.host, base.port) - or (prefix and url.path != prefix and not url.path.startswith(prefix + "/")) + (url.scheme, url.raw_host, url.port) != (base.scheme, base.raw_host, base.port) + or (prefix and path != prefix and not path.startswith(prefix + "/")) ): raise APIError( - f"Refusing to send a request outside {base.scheme}://{base.host}: {url.host}", + f"Refusing to send a request outside {base.scheme}://{base.raw_host.decode()}{prefix}: " + f"{url.raw_host.decode('ascii', 'replace')}", status_code=0, ) @@ -338,7 +379,7 @@ def list_orgs( def get_org(self, org: str) -> dict: """Get a specific organization.""" - return self.get(f"/orgs/{org}/") + return self.get(f"/orgs/{_segment(org)}/") def list_org_repos( self, @@ -351,7 +392,7 @@ def list_org_repos( params: dict[str, Any] = {"limit": limit, "page": page} if repo_type != "all": params["repoType"] = "Source" if repo_type == "source" else "Collection" - return self._get_list(f"/orgs/{org}/repos/", params=params) + return self._get_list(f"/orgs/{_segment(org)}/repos/", params=params) # ── User operations ────────────────────────────────────────── @@ -372,7 +413,7 @@ def list_users( def get_user_detail(self, username: str) -> dict: """Get a specific user by username.""" - return self.get(f"/users/{username}/") + return self.get(f"/users/{_segment(username)}/") def list_user_repos( self, @@ -385,15 +426,15 @@ def list_user_repos( params: dict[str, Any] = {"limit": limit, "page": page} if repo_type != "all": params["repoType"] = "Source" if repo_type == "source" else "Collection" - return self._get_list(f"/users/{username}/repos/", params=params) + return self._get_list(f"/users/{_segment(username)}/repos/", params=params) def list_user_orgs(self, username: str, limit: int = 100) -> dict: """List organizations a user belongs to.""" - return self._get_list(f"/users/{username}/orgs/", params={"limit": limit, "verbose": "true"}) + return self._get_list(f"/users/{_segment(username)}/orgs/", params={"limit": limit, "verbose": "true"}) def get_org_members(self, org: str, limit: int = 100) -> dict: """List members of an organization.""" - return self._get_list(f"/orgs/{org}/members/", params={"limit": limit}) + return self._get_list(f"/orgs/{_segment(org)}/members/", params={"limit": limit}) def search_repos( self, @@ -425,11 +466,11 @@ def search_repos( # Use scoped endpoint when owner is specified if owner and owner_type and owner_type != "all": - ot = owner_type # "orgs" or "users" - endpoint = f"/{ot}/{owner}/repos/" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{_segment(owner)}/repos/" elif owner: # Default to orgs scope - endpoint = f"/orgs/{owner}/repos/" + endpoint = f"/orgs/{_segment(owner)}/repos/" else: # Global search — add owner/ownerType as query params endpoint = "/repos/" @@ -555,12 +596,12 @@ def get_concept( """Get a single concept.""" _validate_owner_type(owner_type) stem = _repo_type_stem("source") - base = f"/{owner_type}/{owner}/{stem}/{source}" + base = f"/{owner_type}/{_segment(owner)}/{stem}/{_segment(source)}" if repo_version: - base = f"{base}/{repo_version}" - endpoint = f"{base}/concepts/{concept_id}/" + base = f"{base}/{_segment(repo_version)}" + endpoint = f"{base}/concepts/{_segment(concept_id)}/" if concept_version: - endpoint = f"{base}/concepts/{concept_id}/{concept_version}/" + endpoint = f"{base}/concepts/{_segment(concept_id)}/{_segment(concept_version)}/" params: dict[str, Any] = {} if include_mappings: @@ -583,7 +624,7 @@ def get_concept_versions( ) -> dict: """List version history for a concept.""" _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/versions/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/versions/" return self._get_list(endpoint, params={"limit": limit, "page": page}) def get_concept_names( @@ -595,7 +636,7 @@ def get_concept_names( ) -> dict: """List names for a concept.""" _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/names/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/names/" return self._get_list(endpoint) def get_concept_descriptions( @@ -607,7 +648,7 @@ def get_concept_descriptions( ) -> dict: """List descriptions for a concept.""" _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/descriptions/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/descriptions/" return self._get_list(endpoint) def get_concept_extras( @@ -619,7 +660,7 @@ def get_concept_extras( ) -> dict: """Get extras (custom attributes) for a concept.""" _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/extras/" return self.get(endpoint) def search_mappings( @@ -701,10 +742,10 @@ def get_mapping( ) -> dict: """Get a single mapping.""" _validate_owner_type(owner_type) - base = f"/{owner_type}/{owner}/sources/{source}" + base = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}" if repo_version: - base = f"{base}/{repo_version}" - return self.get(f"{base}/mappings/{mapping_id}/") + base = f"{base}/{_segment(repo_version)}" + return self.get(f"{base}/mappings/{_segment(mapping_id)}/") def get_mapping_versions( self, @@ -717,7 +758,7 @@ def get_mapping_versions( ) -> dict: """List version history for a mapping.""" _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/versions/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/mappings/{_segment(mapping_id)}/versions/" return self._get_list(endpoint, params={"limit": limit, "page": page}) def cascade( @@ -744,7 +785,7 @@ def cascade( """Execute $cascade operation on a concept.""" endpoint = _build_repo_endpoint( owner_type, owner, repo_type, repo, repo_version, - suffix=f"concepts/{concept_id}/$cascade/" + suffix=f"concepts/{_segment(concept_id)}/$cascade/" ) params: dict[str, Any] = {} @@ -841,7 +882,7 @@ def list_expansions( ) -> dict: """List expansions for a collection version.""" _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/expansions/" + endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/{_segment(collection_version)}/expansions/" return self._get_list(endpoint, params={"verbose": True}) def get_expansion( @@ -857,16 +898,16 @@ def get_expansion( if expansion_id: if collection_version: - endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/expansions/{expansion_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/{_segment(collection_version)}/expansions/{_segment(expansion_id)}/" else: - endpoint = f"/{owner_type}/{owner}/collections/{collection}/expansions/{expansion_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/expansions/{_segment(expansion_id)}/" return self.get(endpoint, params={"verbose": True}) # Look up default expansion from collection version if collection_version: - col_endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/" + col_endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/{_segment(collection_version)}/" else: - col_endpoint = f"/{owner_type}/{owner}/collections/{collection}/" + col_endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/" col_data = self.get(col_endpoint) expansion_url = col_data.get("expansion_url") @@ -893,9 +934,9 @@ def list_expansion_concepts( Uses the version's default expansion when ``expansion_id`` is not given. """ _validate_owner_type(owner_type) - version_endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/" + version_endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/{_segment(collection_version)}/" if expansion_id: - expansion_url = f"{version_endpoint}expansions/{expansion_id}/" + expansion_url = f"{version_endpoint}expansions/{_segment(expansion_id)}/" else: expansion_url = self.get(version_endpoint).get("expansion_url") if not expansion_url: @@ -945,9 +986,9 @@ def list_collection_refs( ) -> dict: """List references in a collection.""" _validate_owner_type(owner_type) - base = f"/{owner_type}/{owner}/collections/{collection}" + base = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}" if collection_version: - base = f"{base}/{collection_version}" + base = f"{base}/{_segment(collection_version)}" endpoint = f"{base}/references/" return self._get_list(endpoint, params={"limit": limit, "page": page}) @@ -982,7 +1023,7 @@ def create_repo( self._require_auth() _validate_owner_type(owner_type) stem = _repo_type_stem(repo_type) - endpoint = f"/{owner_type}/{owner}/{stem}/" + endpoint = f"/{owner_type}/{_segment(owner)}/{stem}/" body: dict[str, Any] = {"id": repo_id, "short_code": repo_id, "name": name} if description: @@ -1060,7 +1101,7 @@ def update_repo_version( self._require_auth() _validate_owner_type(owner_type) stem = _repo_type_stem(repo_type) - endpoint = f"/{owner_type}/{owner}/{stem}/{repo}/{version_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/{stem}/{_segment(repo)}/{_segment(version_id)}/" body = {k: v for k, v in fields.items() if v is not None} return self.patch(endpoint, json=body) @@ -1084,7 +1125,7 @@ def create_concept( if not names: raise ValueError("At least one name is required") - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/" body: dict[str, Any] = { "id": concept_id, "concept_class": concept_class, @@ -1113,7 +1154,7 @@ def update_concept( """Update an existing concept.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/" body = {k: v for k, v in fields.items() if v is not None} if update_comment: body["update_comment"] = update_comment @@ -1145,7 +1186,7 @@ def delete_concept( """Hard-delete a concept (?hardDelete=true). Requires staff token.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/" return self.delete(endpoint, params={"hardDelete": "true"}) def add_concept_name( @@ -1162,7 +1203,7 @@ def add_concept_name( """Add a name/translation to a concept.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/names/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/names/" body: dict[str, Any] = {"name": name, "locale": locale} if name_type: body["name_type"] = name_type @@ -1183,7 +1224,7 @@ def add_concept_description( """Add a description to a concept.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/descriptions/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/descriptions/" body: dict[str, Any] = {"description": description, "locale": locale} if description_type: body["description_type"] = description_type @@ -1201,7 +1242,7 @@ def set_concept_extra( """Set a custom attribute on a concept.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/{key}/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/extras/{_segment(key)}/" return self.put(endpoint, json={key: value}) def delete_concept_extra( @@ -1215,7 +1256,7 @@ def delete_concept_extra( """Delete a custom attribute from a concept.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/{key}/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/concepts/{_segment(concept_id)}/extras/{_segment(key)}/" return self.delete(endpoint) def create_mapping( @@ -1244,7 +1285,7 @@ def create_mapping( if not to_concept_url and not (to_source_url and to_concept_code): raise ValueError("to_concept requires either to_concept_url or (to_source_url + to_concept_code)") - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/mappings/" body: dict[str, Any] = {"map_type": map_type} if from_concept_url: @@ -1282,7 +1323,7 @@ def update_mapping( """Update an existing mapping.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/mappings/{_segment(mapping_id)}/" body = {k: v for k, v in fields.items() if v is not None} if update_comment: body["update_comment"] = update_comment @@ -1314,7 +1355,7 @@ def delete_mapping( """Hard-delete a mapping (?hardDelete=true). Requires staff token.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(source)}/mappings/{_segment(mapping_id)}/" return self.delete(endpoint, params={"hardDelete": "true"}) def add_collection_ref( @@ -1328,7 +1369,7 @@ def add_collection_ref( """Add references to a collection.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/references/" + endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/references/" body: dict[str, Any] = {"data": {"expressions": expressions}} params: dict[str, Any] = {} if cascade: @@ -1345,7 +1386,7 @@ def remove_collection_ref( """Remove references from a collection.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/references/" + endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/references/" # DELETE with body response = self.client.request( "DELETE", endpoint, json={"references": expressions} @@ -1366,7 +1407,7 @@ def create_expansion( """Create/trigger an expansion for a collection version.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/{version}/expansions/" + endpoint = f"/{owner_type}/{_segment(owner)}/collections/{_segment(collection)}/{_segment(version)}/expansions/" body = {k: v for k, v in params.items() if v is not None} return self.post(endpoint, json=body or None) @@ -1386,7 +1427,7 @@ def clone( """Clone concepts using $clone operation.""" self._require_auth() _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{dest_source}/$clone/" + endpoint = f"/{owner_type}/{_segment(owner)}/sources/{_segment(dest_source)}/$clone/" body: dict[str, Any] = {"expressions": expressions} params: dict[str, Any] = {} @@ -1416,7 +1457,7 @@ def set_repo_extra( ) -> dict: """Set a custom attribute on a repository.""" self._require_auth() - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix=f"extras/{key}/") + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix=f"extras/{_segment(key)}/") return self.put(endpoint, json={key: value}) def delete_repo_extra( @@ -1431,7 +1472,7 @@ def delete_repo_extra( self._require_auth() _validate_owner_type(owner_type) stem = _repo_type_stem(repo_type) - return self.delete(f"/{owner_type}/{owner}/{stem}/{repo}/extras/{key}/") + return self.delete(f"/{owner_type}/{_segment(owner)}/{stem}/{_segment(repo)}/extras/{_segment(key)}/") def get_repo_extras( self, @@ -1447,12 +1488,12 @@ def get_repo_extras( def add_org_member(self, org: str, username: str) -> dict: """Add a member to an organization.""" self._require_auth() - return self.put(f"/orgs/{org}/members/{username}/") + return self.put(f"/orgs/{_segment(org)}/members/{_segment(username)}/") def remove_org_member(self, org: str, username: str) -> dict: """Remove a member from an organization.""" self._require_auth() - return self.delete(f"/orgs/{org}/members/{username}/") + return self.delete(f"/orgs/{_segment(org)}/members/{_segment(username)}/") # ── Task operations ───────────────────────────────────────────── @@ -1472,7 +1513,7 @@ def list_tasks( def get_task(self, task_id: str) -> dict: """Get details of a specific task.""" self._require_auth() - return self.get(f"/tasks/{task_id}/") + return self.get(f"/tasks/{_segment(task_id)}/") # ── Organization CRUD ─────────────────────────────────────────── @@ -1506,7 +1547,7 @@ def create_org( def delete_org(self, org: str) -> dict: """Delete an organization.""" self._require_auth() - return self.delete(f"/orgs/{org}/") + return self.delete(f"/orgs/{_segment(org)}/") # ── Repository delete ─────────────────────────────────────────── @@ -1580,7 +1621,7 @@ def import_file( endpoint = "/importers/bulk-import/" if queue: - endpoint = f"/importers/bulk-import/{queue}/" + endpoint = f"/importers/bulk-import/{_segment(queue)}/" params = {"update_if_exists": "true" if update_if_exists else "false"} @@ -1612,7 +1653,7 @@ def import_list(self, queue: str | None = None) -> Any: self._require_auth() endpoint = "/importers/bulk-import/" if queue: - endpoint = f"/importers/bulk-import/{queue}/" + endpoint = f"/importers/bulk-import/{_segment(queue)}/" return self.get(endpoint) def import_status(self, task_id: str) -> Any: diff --git a/tests/test_client.py b/tests/test_client.py index 4e0bf7a..e56ba98 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -231,3 +231,192 @@ def test_relative_path_cannot_leave_base_url_prefix(httpx_mock): def test_same_origin_absolute_url_is_allowed(httpx_mock): httpx_mock.add_response(url=f"{BASE}/x/", json={"ok": True}) assert OCLAPIClient(base_url=BASE).get(f"{BASE}/x/") == {"ok": True} + + +# ── Identifiers in paths, and the origin check (ocl_online#553) ────────── + + +@pytest.fixture +def no_retry_wait(monkeypatch): + for name in ("get", "post", "put", "patch", "delete", "_get_list"): + monkeypatch.setattr(getattr(OCLAPIClient, name).retry, "wait", wait_none()) + + +def test_an_identifier_is_encoded_as_one_path_segment(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=BASE).get_concept("CIEL", "CIEL", "A B+50%") + assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/A%20B%2B50%25/" + + +def test_a_percent_escape_in_an_identifier_is_sent_literally(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=BASE).get_concept("CIEL", "CIEL", "a%2Fb") + assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/a%252Fb/" + + +def test_identifiers_are_encoded_under_a_base_url_prefix(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url="https://api.example.test/ocl-api").get_concept("X", "S", "a b") + assert httpx_mock.get_requests()[0].url.raw_path == b"/ocl-api/orgs/X/sources/S/concepts/a%20b/" + + +# Every method that puts a caller's value in a path, with that value set to v. +PATH_BUILDERS = { + "get_org": lambda c, v: c.get_org(v), + "list_org_repos": lambda c, v: c.list_org_repos(v), + "get_user_detail": lambda c, v: c.get_user_detail(v), + "list_user_repos": lambda c, v: c.list_user_repos(v), + "list_user_orgs": lambda c, v: c.list_user_orgs(v), + "get_org_members": lambda c, v: c.get_org_members(v), + "search_repos owner": lambda c, v: c.search_repos(owner=v), + "search_repos scoped owner": lambda c, v: c.search_repos(owner=v, owner_type="users"), + "get_repo": lambda c, v: c.get_repo("X", v), + "get_repo version": lambda c, v: c.get_repo("X", "R", repo_version=v), + "get_repo_versions": lambda c, v: c.get_repo_versions(v, "R"), + "search_concepts": lambda c, v: c.search_concepts(owner="X", repo=v), + "get_concept": lambda c, v: c.get_concept("X", "S", v), + "get_concept repo version": lambda c, v: c.get_concept("X", "S", "C", repo_version=v), + "get_concept concept version": lambda c, v: c.get_concept("X", "S", "C", concept_version=v), + "get_concept_versions": lambda c, v: c.get_concept_versions("X", "S", v), + "get_concept_names": lambda c, v: c.get_concept_names("X", "S", v), + "get_concept_descriptions": lambda c, v: c.get_concept_descriptions("X", "S", v), + "get_concept_extras": lambda c, v: c.get_concept_extras("X", v, "C"), + "search_mappings": lambda c, v: c.search_mappings(owner="X", repo=v), + "get_mapping": lambda c, v: c.get_mapping("X", "S", v), + "get_mapping repo version": lambda c, v: c.get_mapping("X", "S", "M", repo_version=v), + "get_mapping_versions": lambda c, v: c.get_mapping_versions("X", "S", v), + "cascade": lambda c, v: c.cascade("X", "S", v), + "list_expansions": lambda c, v: c.list_expansions("X", "C", v), + "get_expansion": lambda c, v: c.get_expansion("X", "C", expansion_id=v), + "get_expansion in a version": lambda c, v: c.get_expansion("X", "C", collection_version="v1", expansion_id=v), + "get_expansion default": lambda c, v: c.get_expansion("X", "C", collection_version=v), + "list_expansion_concepts": lambda c, v: c.list_expansion_concepts("X", "C", "v1", expansion_id=v), + "list_expansion_concepts version": lambda c, v: c.list_expansion_concepts("X", "C", v, expansion_id="e1"), + "list_collection_refs": lambda c, v: c.list_collection_refs("X", v), + "list_collection_refs version": lambda c, v: c.list_collection_refs("X", "C", collection_version=v), + "create_repo": lambda c, v: c.create_repo(v, "R", "Name"), + "update_repo": lambda c, v: c.update_repo("X", v, name="n"), + "create_repo_version": lambda c, v: c.create_repo_version("X", v, "v1"), + "update_repo_version": lambda c, v: c.update_repo_version("X", "R", v, description="d"), + "create_concept": lambda c, v: c.create_concept("X", v, "C", "Misc", [{"name": "n", "locale": "en"}]), + "update_concept": lambda c, v: c.update_concept("X", "S", v, retired=False), + "retire_concept": lambda c, v: c.retire_concept("X", "S", v), + "delete_concept": lambda c, v: c.delete_concept("X", "S", v), + "add_concept_name": lambda c, v: c.add_concept_name("X", "S", v, "n", "en"), + "add_concept_description": lambda c, v: c.add_concept_description("X", "S", v, "d", "en"), + "set_concept_extra": lambda c, v: c.set_concept_extra("X", "S", "C", v, 1), + "delete_concept_extra": lambda c, v: c.delete_concept_extra("X", "S", "C", v), + "create_mapping": lambda c, v: c.create_mapping("X", v, "SAME-AS", from_concept_url="/a/", to_concept_url="/b/"), + "update_mapping": lambda c, v: c.update_mapping("X", "S", v, retired=False), + "retire_mapping": lambda c, v: c.retire_mapping("X", "S", v), + "delete_mapping": lambda c, v: c.delete_mapping("X", "S", v), + "add_collection_ref": lambda c, v: c.add_collection_ref("X", v, ["/a/"]), + "remove_collection_ref": lambda c, v: c.remove_collection_ref("X", v, ["/a/"]), + "create_expansion": lambda c, v: c.create_expansion("X", "C", v), + "clone": lambda c, v: c.clone("X", v, ["/a/"]), + "set_repo_extra": lambda c, v: c.set_repo_extra("X", "R", v, 1), + "delete_repo_extra": lambda c, v: c.delete_repo_extra("X", "R", v), + "get_repo_extras": lambda c, v: c.get_repo_extras("X", v), + "add_org_member": lambda c, v: c.add_org_member("X", v), + "remove_org_member": lambda c, v: c.remove_org_member("X", v), + "get_task": lambda c, v: c.get_task(v), + "delete_org": lambda c, v: c.delete_org(v), + "delete_repo": lambda c, v: c.delete_repo("X", v), + "import_file": lambda c, v: c.import_file("unused.json", queue=v), + "import_list": lambda c, v: c.import_list(queue=v), + "export_status": lambda c, v: c.export_status("X", "R", v), + "export_create": lambda c, v: c.export_create("X", "R", v), + "export_delete": lambda c, v: c.export_delete("X", "R", v), + "export_download": lambda c, v: c.export_download("X", "R", v), +} + + +@pytest.mark.parametrize("value", ["..", ".", "a/b"]) +@pytest.mark.parametrize("call", PATH_BUILDERS.values(), ids=PATH_BUILDERS.keys()) +def test_an_identifier_that_would_change_the_endpoint_is_refused(httpx_mock, no_retry_wait, call, value): + with pytest.raises(APIError) as err: + call(OCLAPIClient(base_url=BASE, token="t"), value) + assert err.value.status_code == 0 + assert httpx_mock.get_requests() == [] + + +def test_an_empty_identifier_is_refused(httpx_mock, no_retry_wait): + client = OCLAPIClient(base_url=BASE, token="t") + for call in (lambda: client.get_org(""), lambda: client.update_repo_version("X", "R", "", description="d")): + with pytest.raises(APIError): + call() + assert httpx_mock.get_requests() == [] + + +def test_search_repos_refuses_an_unknown_owner_type(httpx_mock): + with pytest.raises(ValueError): + OCLAPIClient(base_url=BASE).search_repos(owner="X", owner_type="../admin") + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("path", ["%2e%2e/other/", "a%2F..%2F..%2Fother/", "..%5C..%5Cother/"]) +def test_an_encoded_dot_segment_cannot_leave_base_url_prefix(httpx_mock, path): + with pytest.raises(APIError): + OCLAPIClient(base_url="https://api.example.test/ocl-api").get(f"https://api.example.test/ocl-api/{path}") + assert httpx_mock.get_requests() == [] + + +def test_an_encoded_backslash_inside_the_prefix_is_allowed(httpx_mock): + url = "https://api.example.test/ocl-api/orgs/X/sources/S/concepts/A%5CB/" + httpx_mock.add_response(url=url, json={"ok": True}) + assert OCLAPIClient(base_url="https://api.example.test/ocl-api").get(url) == {"ok": True} + + +def test_a_malformed_international_host_is_refused(httpx_mock): + with pytest.raises(APIError): + OCLAPIClient(base_url=BASE, token="t").get("https://xn--a.example/x/") + assert httpx_mock.get_requests() == [] + + +def test_an_international_host_matches_its_punycode_form(httpx_mock): + httpx_mock.add_response(url="https://xn--bcher-kva.example/x/", json={"ok": True}) + assert OCLAPIClient(base_url="https://bücher.example").get("https://xn--bcher-kva.example/x/") == {"ok": True} + + +def test_a_url_with_credentials_is_refused(httpx_mock, no_retry_wait): + with pytest.raises(APIError) as err: + OCLAPIClient(base_url=BASE, token="t").get("https://user:pw@api.example.test/x/") + assert "pw" not in str(err.value) + assert httpx_mock.get_requests() == [] + + +def test_ocl_concept_get_encodes_special_characters(httpx_mock): + from ocl_cli.main import cli + + httpx_mock.add_response(json={"id": "A B+50%"}) + with patch("ocl_cli.main.CLIConfig.load", return_value=_FakeConfig()): + result = CliRunner().invoke(cli, ["--json", "concept", "get", "CIEL", "CIEL", "A B+50%"]) + assert result.exit_code == 0, result.output + assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/A%20B%2B50%25/" + + +def test_recreate_collection_refuses_an_export_whose_id_would_change_the_endpoint(httpx_mock, no_retry_wait, tmp_path): + from ocl_cli.main import cli + + export = tmp_path / "export.json" + export.write_text(json.dumps({"collection": {"id": ".."}, "references": []})) + with patch("ocl_cli.main.CLIConfig.load", return_value=_FakeConfig()): + result = CliRunner().invoke(cli, [ + "tool", "recreate-collection", "--input", str(export), "--org", "PIH", + "--output", str(tmp_path / "refs.json"), "--no-dry-run", + ]) + assert httpx_mock.get_requests() == [] + assert result.exit_code == 1 + assert "Invalid identifier" in result.output + + +def test_reserved_and_non_ascii_characters_are_encoded(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=BASE).get_concept("CIEL", "CIEL", "é&:[] x") + assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/%C3%A9%26%3A%5B%5D%20x/" + + +def test_get_still_takes_a_whole_path(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=BASE).get("/orgs/CIEL/sources/CIEL/concepts/a b&c/") + assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/a%20b&c/" From dc102201ebe02ee700a7163af1293696d5a0fe10 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 11:52:31 -0400 Subject: [PATCH 2/7] OpenConceptLab/ocl_online#553 | Prefix check holds for every reading of the path; exports need a version - A base_url path prefix must hold for the path as sent, with dot segments resolved, and with backslashes read as slashes. Reading backslashes as slashes alone let a path outside the prefix pass. - The export methods require a version: an empty one would make the path name a version called "export". - An invalid host in a redirect's Location raises APIError too: httpx builds the next request even when it doesn't follow redirects. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 27 +++++++++++++++++---------- tests/test_client.py | 20 ++++++++++++++++++++ 2 files changed, 37 insertions(+), 10 deletions(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 3f511ca..e36a800 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -65,9 +65,9 @@ def _segment(value: Any) -> str: def _resolve_dot_segments(path: str) -> str: - """The path as a server that decodes and normalizes it would route it.""" + """The path with its ``.`` and ``..`` segments resolved, as a normalizing server routes it.""" parts: list[str] = [] - for part in path.replace("\\", "/").split("/"): + for part in path.split("/"): if part == "..": if parts: parts.pop() @@ -76,6 +76,10 @@ def _resolve_dot_segments(path: str) -> str: return "/" + "/".join(parts) +def _within(path: str, prefix: str) -> bool: + return path == prefix or path.startswith(prefix + "/") + + def _validate_owner_type(owner_type: str) -> None: if owner_type not in ("users", "orgs"): raise ValueError(f"owner_type must be 'users' or 'orgs', got '{owner_type}'") @@ -99,11 +103,12 @@ def _build_repo_endpoint( class _HTTPClient(httpx.Client): - def build_request(self, *args: Any, **kwargs: Any) -> httpx.Request: + def request(self, *args: Any, **kwargs: Any) -> httpx.Response: + # httpx decodes a URL's host when it builds a request, a redirect's next one included. try: - return super().build_request(*args, **kwargs) + return super().request(*args, **kwargs) except UnicodeError as e: # idna's error for a host that isn't a valid domain name - raise APIError(f"Refusing to send a request to an invalid host: {e}", status_code=0) from e + raise APIError(f"Refusing a URL whose host isn't a valid domain name: {e}", status_code=0) from e class OCLAPIClient: @@ -157,17 +162,18 @@ def __init__( def _check_destination(self, request: httpx.Request) -> None: """Refuse any request outside the API's origin and path: the token goes only there. - Hosts are compared in their encoded (punycode) form, and the path as a server - that decodes and normalizes it would route it. + Hosts are compared in their encoded (punycode) form. A path prefix must hold + however a server routes the decoded path: as sent, with dot segments resolved, + or with backslashes read as slashes too. """ base, url = self._base, request.url if url.userinfo: raise APIError("Refusing to send a request with credentials in its URL", status_code=0) prefix = base.path.rstrip("/") - path = _resolve_dot_segments(url.path) + readings = (url.path, _resolve_dot_segments(url.path), _resolve_dot_segments(url.path.replace("\\", "/"))) if ( (url.scheme, url.raw_host, url.port) != (base.scheme, base.raw_host, base.port) - or (prefix and path != prefix and not path.startswith(prefix + "/")) + or (prefix and not all(_within(path, prefix) for path in readings)) ): raise APIError( f"Refusing to send a request outside {base.scheme}://{base.raw_host.decode()}{prefix}: " @@ -1697,8 +1703,9 @@ def _export_endpoint( self, owner: str, repo: str, version: str, owner_type: str, repo_type: str, ) -> str: + # The version is required: without it the path would name a version called "export". return _build_repo_endpoint( - owner_type, owner, repo_type, repo, version, "export/" + owner_type, owner, repo_type, repo, suffix=f"{_segment(version)}/export/" ) def export_status( diff --git a/tests/test_client.py b/tests/test_client.py index e56ba98..7b6bf9b 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -420,3 +420,23 @@ def test_get_still_takes_a_whole_path(httpx_mock): httpx_mock.add_response(json={}) OCLAPIClient(base_url=BASE).get("/orgs/CIEL/sources/CIEL/concepts/a b&c/") assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/a%20b&c/" + + +def test_a_path_outside_the_prefix_that_normalizes_inside_is_refused(httpx_mock): + with pytest.raises(APIError): + OCLAPIClient(base_url="https://api.example.test/ocl-api").get("https://api.example.test/other/..%5Cocl-api/x/") + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("method", ["export_status", "export_create", "export_delete", "export_download"]) +def test_an_export_needs_a_version(httpx_mock, method): + with pytest.raises(APIError): + getattr(OCLAPIClient(base_url=BASE, token="t"), method)("X", "R", "") + assert httpx_mock.get_requests() == [] + + +@pytest.mark.parametrize("follow_redirects", [False, True]) +def test_a_redirect_to_a_malformed_host_raises_api_error(httpx_mock, follow_redirects): + httpx_mock.add_response(url=f"{BASE}/x/", status_code=302, headers={"Location": "https://xn--a.example/y/"}) + with pytest.raises(APIError): + OCLAPIClient(base_url=BASE, token="t").request("GET", "/x/", follow_redirects=follow_redirects) From d03708bc0e4a7448866d2cca5d43253e4e8c815c Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 11:58:08 -0400 Subject: [PATCH 3/7] OpenConceptLab/ocl_online#553 | Cascade lookups only for expressions that name their own endpoint fetch_cascade_children takes expressions from export data. It now returns no children, without a request, for an expression with a "." or ".." segment (encoded or not), a "?" or a "#". Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 7 ++++++- tests/test_client.py | 20 ++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index e36a800..8c55982 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -6,7 +6,7 @@ import platform import sys from typing import Any, Optional -from urllib.parse import quote +from urllib.parse import quote, unquote import httpx from tenacity import ( @@ -1671,6 +1671,11 @@ def import_status(self, task_id: str) -> Any: def fetch_cascade_children(self, concept_url: str) -> set[str]: """Fetch child concept URLs via cascade for a given concept expression.""" + # The expression comes from data, so it may only name its own cascade endpoint. + if "?" in concept_url or "#" in concept_url or any( + part in (".", "..") for part in unquote(concept_url).split("/") + ): + return set() url = concept_url.rstrip("/") + "/cascade/" params = { "method": "sourcetoconcepts", diff --git a/tests/test_client.py b/tests/test_client.py index 7b6bf9b..a6ece0a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -440,3 +440,23 @@ def test_a_redirect_to_a_malformed_host_raises_api_error(httpx_mock, follow_redi httpx_mock.add_response(url=f"{BASE}/x/", status_code=302, headers={"Location": "https://xn--a.example/y/"}) with pytest.raises(APIError): OCLAPIClient(base_url=BASE, token="t").request("GET", "/x/", follow_redirects=follow_redirects) + + +@pytest.mark.parametrize("expression", [ + "/orgs/X/sources/S/concepts/../", "/orgs/X/sources/S/concepts/./", "/orgs/X/sources/S/concepts/%2e%2e/", + "/orgs/X/sources/S/concepts/1?x=/", "/orgs/X/sources/S/concepts/1#/", +]) +def test_cascade_children_of_an_expression_that_would_change_the_endpoint_are_not_fetched( + httpx_mock, no_retry_wait, expression +): + assert OCLAPIClient(base_url=BASE, token="t").fetch_cascade_children(expression) == set() + assert httpx_mock.get_requests() == [] + + +def test_cascade_children_are_fetched_for_a_concept_expression(httpx_mock): + httpx_mock.add_response( + url=re.compile(rf"{re.escape(BASE)}/orgs/X/sources/S/concepts/1/cascade/\?.*"), + json={"entry": [{"type": "Concept", "url": "/orgs/X/sources/S/concepts/2/"}]}, + ) + children = OCLAPIClient(base_url=BASE, token="t").fetch_cascade_children("/orgs/X/sources/S/concepts/1/") + assert children == {"/orgs/X/sources/S/concepts/2/"} From 12aa8b29e66a1101e4d5df836384c3adfde49681 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 11:59:33 -0400 Subject: [PATCH 4/7] OpenConceptLab/ocl_online#553 | Simplify: inline the prefix rule, merge two encoding tests Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 6 +----- tests/test_client.py | 12 ++++-------- 2 files changed, 5 insertions(+), 13 deletions(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 8c55982..13e742b 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -76,10 +76,6 @@ def _resolve_dot_segments(path: str) -> str: return "/" + "/".join(parts) -def _within(path: str, prefix: str) -> bool: - return path == prefix or path.startswith(prefix + "/") - - def _validate_owner_type(owner_type: str) -> None: if owner_type not in ("users", "orgs"): raise ValueError(f"owner_type must be 'users' or 'orgs', got '{owner_type}'") @@ -173,7 +169,7 @@ def _check_destination(self, request: httpx.Request) -> None: readings = (url.path, _resolve_dot_segments(url.path), _resolve_dot_segments(url.path.replace("\\", "/"))) if ( (url.scheme, url.raw_host, url.port) != (base.scheme, base.raw_host, base.port) - or (prefix and not all(_within(path, prefix) for path in readings)) + or (prefix and not all(path == prefix or path.startswith(prefix + "/") for path in readings)) ): raise APIError( f"Refusing to send a request outside {base.scheme}://{base.raw_host.decode()}{prefix}: " diff --git a/tests/test_client.py b/tests/test_client.py index a6ece0a..d0c8426 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -244,8 +244,10 @@ def no_retry_wait(monkeypatch): def test_an_identifier_is_encoded_as_one_path_segment(httpx_mock): httpx_mock.add_response(json={}) - OCLAPIClient(base_url=BASE).get_concept("CIEL", "CIEL", "A B+50%") - assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/A%20B%2B50%25/" + OCLAPIClient(base_url=BASE).get_concept("CIEL", "CIEL", "A B+50%é&:[]") + assert httpx_mock.get_requests()[0].url.raw_path == ( + b"/orgs/CIEL/sources/CIEL/concepts/A%20B%2B50%25%C3%A9%26%3A%5B%5D/" + ) def test_a_percent_escape_in_an_identifier_is_sent_literally(httpx_mock): @@ -410,12 +412,6 @@ def test_recreate_collection_refuses_an_export_whose_id_would_change_the_endpoin assert "Invalid identifier" in result.output -def test_reserved_and_non_ascii_characters_are_encoded(httpx_mock): - httpx_mock.add_response(json={}) - OCLAPIClient(base_url=BASE).get_concept("CIEL", "CIEL", "é&:[] x") - assert httpx_mock.get_requests()[0].url.raw_path == b"/orgs/CIEL/sources/CIEL/concepts/%C3%A9%26%3A%5B%5D%20x/" - - def test_get_still_takes_a_whole_path(httpx_mock): httpx_mock.add_response(json={}) OCLAPIClient(base_url=BASE).get("/orgs/CIEL/sources/CIEL/concepts/a b&c/") From 5207d9c0846d0796ee43fd3edd236caef5309889 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 12:13:52 -0400 Subject: [PATCH 5/7] OpenConceptLab/ocl_online#553 | Prefix check: the raw path, and paths decoded until stable A base_url path prefix must now hold for the raw path as sent, and for a reading of the decoded path that is decoded until it stops changing, reads backslashes as slashes and drops ";" parameters, besides the readings already checked. Gateways that route on the raw path, decode twice or honour path parameters can no longer take a request outside the prefix. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 32 ++++++++++++++++++++++++++++---- tests/test_client.py | 20 ++++++++++++++++++++ 2 files changed, 48 insertions(+), 4 deletions(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 13e742b..0ec6d70 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -76,6 +76,28 @@ def _resolve_dot_segments(path: str) -> str: return "/" + "/".join(parts) +def _path_readings(path: str) -> tuple[str, ...]: + """Ways a server or gateway may route a decoded path. + + As is; with dot segments resolved; with backslashes read as slashes; and + decoded until it stops changing, with ``;`` parameters dropped from each segment. + """ + deep = path + while (decoded := unquote(deep)) != deep: + deep = decoded + deep = "/".join(part.split(";")[0] for part in deep.replace("\\", "/").split("/")) + return ( + path, + _resolve_dot_segments(path), + _resolve_dot_segments(path.replace("\\", "/")), + _resolve_dot_segments(deep), + ) + + +def _within(path: str, prefix: str) -> bool: + return path == prefix or path.startswith(prefix + "/") + + def _validate_owner_type(owner_type: str) -> None: if owner_type not in ("users", "orgs"): raise ValueError(f"owner_type must be 'users' or 'orgs', got '{owner_type}'") @@ -159,17 +181,19 @@ def _check_destination(self, request: httpx.Request) -> None: """Refuse any request outside the API's origin and path: the token goes only there. Hosts are compared in their encoded (punycode) form. A path prefix must hold - however a server routes the decoded path: as sent, with dot segments resolved, - or with backslashes read as slashes too. + for the raw path and for every reading of the decoded one (``_path_readings``). """ base, url = self._base, request.url if url.userinfo: raise APIError("Refusing to send a request with credentials in its URL", status_code=0) prefix = base.path.rstrip("/") - readings = (url.path, _resolve_dot_segments(url.path), _resolve_dot_segments(url.path.replace("\\", "/"))) + raw_prefix = base.raw_path.decode("ascii").rstrip("/") + raw_path = url.raw_path.decode("ascii").split("?")[0] if ( (url.scheme, url.raw_host, url.port) != (base.scheme, base.raw_host, base.port) - or (prefix and not all(path == prefix or path.startswith(prefix + "/") for path in readings)) + or (prefix and not ( + _within(raw_path, raw_prefix) and all(_within(path, prefix) for path in _path_readings(url.path)) + )) ): raise APIError( f"Refusing to send a request outside {base.scheme}://{base.raw_host.decode()}{prefix}: " diff --git a/tests/test_client.py b/tests/test_client.py index d0c8426..2a2d169 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -456,3 +456,23 @@ def test_cascade_children_are_fetched_for_a_concept_expression(httpx_mock): ) children = OCLAPIClient(base_url=BASE, token="t").fetch_cascade_children("/orgs/X/sources/S/concepts/1/") assert children == {"/orgs/X/sources/S/concepts/2/"} + + +@pytest.mark.parametrize("path", ["/%6Fcl-api/x/", "/ocl-api%2Fx/", "/ocl-api/..;/x/", "/ocl-api/%252e%252e/x/"]) +def test_a_path_outside_the_prefix_under_any_reading_is_refused(httpx_mock, path): + with pytest.raises(APIError): + OCLAPIClient(base_url="https://api.example.test/ocl-api").get(f"https://api.example.test{path}") + assert httpx_mock.get_requests() == [] + + +def test_a_redirect_outside_the_raw_prefix_is_refused(httpx_mock): + httpx_mock.add_response(url="https://api.example.test/ocl-api/x/", status_code=302, headers={"Location": "/%6Fcl-api/y/"}) + with pytest.raises(APIError): + OCLAPIClient(base_url="https://api.example.test/ocl-api", token="t").request("GET", "/x/", follow_redirects=True) + assert len(httpx_mock.get_requests()) == 1 + + +def test_an_encoded_prefix_and_a_semicolon_in_an_identifier_still_work(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url="https://api.example.test/my api").get_concept("X", "S", "a;b") + assert httpx_mock.get_requests()[0].url.raw_path == b"/my%20api/orgs/X/sources/S/concepts/a%3Bb/" From 9e2751d8aefcc8c27be3d7ecc79b7b74d655b06f Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 12:19:21 -0400 Subject: [PATCH 6/7] OpenConceptLab/ocl_online#553 | Prefix readings: compare like with like, bound the decoding - Each reading of the request path is compared with the same reading of the base_url prefix, so a prefix that contains ";" or a literal "%25" works again. - Decoding stops after four passes; a path still changing then is refused, which bounds the work an untrusted URL can cause. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 37 +++++++++++++++++++++++------------- tests/test_client.py | 13 +++++++++++++ 2 files changed, 37 insertions(+), 13 deletions(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 0ec6d70..68bc874 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -76,15 +76,21 @@ def _resolve_dot_segments(path: str) -> str: return "/" + "/".join(parts) -def _path_readings(path: str) -> tuple[str, ...]: +def _path_readings(path: str) -> Optional[tuple[str, ...]]: """Ways a server or gateway may route a decoded path. As is; with dot segments resolved; with backslashes read as slashes; and decoded until it stops changing, with ``;`` parameters dropped from each segment. + None when the path is still changing after a few decoding passes. """ deep = path - while (decoded := unquote(deep)) != deep: + for _ in range(4): + decoded = unquote(deep) + if decoded == deep: + break deep = decoded + else: + return None deep = "/".join(part.split(";")[0] for part in deep.replace("\\", "/").split("/")) return ( path, @@ -181,26 +187,31 @@ def _check_destination(self, request: httpx.Request) -> None: """Refuse any request outside the API's origin and path: the token goes only there. Hosts are compared in their encoded (punycode) form. A path prefix must hold - for the raw path and for every reading of the decoded one (``_path_readings``). + for the raw path, and for each reading of the decoded path against the same + reading of the prefix (``_path_readings``). """ base, url = self._base, request.url if url.userinfo: raise APIError("Refusing to send a request with credentials in its URL", status_code=0) - prefix = base.path.rstrip("/") - raw_prefix = base.raw_path.decode("ascii").rstrip("/") - raw_path = url.raw_path.decode("ascii").split("?")[0] - if ( - (url.scheme, url.raw_host, url.port) != (base.scheme, base.raw_host, base.port) - or (prefix and not ( - _within(raw_path, raw_prefix) and all(_within(path, prefix) for path in _path_readings(url.path)) - )) - ): + if (url.scheme, url.raw_host, url.port) != (base.scheme, base.raw_host, base.port) or not self._within_prefix(url): raise APIError( - f"Refusing to send a request outside {base.scheme}://{base.raw_host.decode()}{prefix}: " + f"Refusing to send a request outside {base.scheme}://{base.raw_host.decode()}{base.path.rstrip('/')}: " f"{url.raw_host.decode('ascii', 'replace')}", status_code=0, ) + def _within_prefix(self, url: httpx.URL) -> bool: + prefix = self._base.path.rstrip("/") + if not prefix: + return True + raw_prefix = self._base.raw_path.decode("ascii").rstrip("/") + if not _within(url.raw_path.decode("ascii").split("?")[0], raw_prefix): + return False + readings, prefix_readings = _path_readings(url.path), _path_readings(prefix) + return bool(readings and prefix_readings) and all( + _within(path, p) for path, p in zip(readings, prefix_readings) + ) + def close(self): self.client.close() diff --git a/tests/test_client.py b/tests/test_client.py index 2a2d169..56cf582 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -476,3 +476,16 @@ def test_an_encoded_prefix_and_a_semicolon_in_an_identifier_still_work(httpx_moc httpx_mock.add_response(json={}) OCLAPIClient(base_url="https://api.example.test/my api").get_concept("X", "S", "a;b") assert httpx_mock.get_requests()[0].url.raw_path == b"/my%20api/orgs/X/sources/S/concepts/a%3Bb/" + + +@pytest.mark.parametrize("base", ["https://api.example.test/api;v=1", "https://api.example.test/api%2520v1"]) +def test_a_prefix_with_a_parameter_or_a_literal_escape_still_works(httpx_mock, base): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=base).get_org("X") + assert len(httpx_mock.get_requests()) == 1 + + +def test_a_path_that_keeps_decoding_is_refused(httpx_mock): + with pytest.raises(APIError): + OCLAPIClient(base_url="https://api.example.test/api").get("https://api.example.test/api/%" + "25" * 50 + "61/") + assert httpx_mock.get_requests() == [] From b3b7cf88a7ad53780e7174ee40b1f850f27b5a8e Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 12:23:32 -0400 Subject: [PATCH 7/7] OpenConceptLab/ocl_online#553 | Prefix readings at every decoding depth The path and the base_url prefix are now compared at each of five decoding depths, not only as sent and fully decoded, so a path that leaves the prefix at an intermediate depth is refused. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 36 +++++++++++++++++------------------- tests/test_client.py | 7 +++++++ 2 files changed, 24 insertions(+), 19 deletions(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 68bc874..8c1628a 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -76,28 +76,26 @@ def _resolve_dot_segments(path: str) -> str: return "/" + "/".join(parts) -def _path_readings(path: str) -> Optional[tuple[str, ...]]: +def _path_readings(path: str) -> Optional[list[str]]: """Ways a server or gateway may route a decoded path. - As is; with dot segments resolved; with backslashes read as slashes; and - decoded until it stops changing, with ``;`` parameters dropped from each segment. - None when the path is still changing after a few decoding passes. + At each of five decoding depths: as is; with dot segments resolved; with + backslashes read as slashes; and with ``;`` parameters dropped from each segment + too. Every path gives the same number of readings, so two paths can be compared + reading by reading. None when the path is still changing after the last depth. """ - deep = path - for _ in range(4): - decoded = unquote(deep) - if decoded == deep: - break - deep = decoded - else: - return None - deep = "/".join(part.split(";")[0] for part in deep.replace("\\", "/").split("/")) - return ( - path, - _resolve_dot_segments(path), - _resolve_dot_segments(path.replace("\\", "/")), - _resolve_dot_segments(deep), - ) + readings: list[str] = [] + for _ in range(5): + slashed = path.replace("\\", "/") + unparameterized = "/".join(part.split(";")[0] for part in slashed.split("/")) + readings += [ + path, + _resolve_dot_segments(path), + _resolve_dot_segments(slashed), + _resolve_dot_segments(unparameterized), + ] + path, previous = unquote(path), path + return readings if path == previous else None def _within(path: str, prefix: str) -> bool: diff --git a/tests/test_client.py b/tests/test_client.py index 56cf582..c81162f 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -489,3 +489,10 @@ def test_a_path_that_keeps_decoding_is_refused(httpx_mock): with pytest.raises(APIError): OCLAPIClient(base_url="https://api.example.test/api").get("https://api.example.test/api/%" + "25" * 50 + "61/") assert httpx_mock.get_requests() == [] + + +def test_a_path_that_leaves_the_prefix_at_an_intermediate_decoding_is_refused(httpx_mock): + base = "https://api.example.test/api%2520v1" + with pytest.raises(APIError): + OCLAPIClient(base_url=base).get(f"{base}/%252e%252e/api%252520v1/x/") + assert httpx_mock.get_requests() == []