From 4ca84fc5dd1f24f495d06103edca1e7c82a98232 Mon Sep 17 00:00:00 2001 From: Ivan Date: Fri, 1 May 2026 04:13:55 -0500 Subject: [PATCH] feat(community): implement URL validation and fetch handling for community directory to prevent SSRF vulnerabilities; add tests for validation and fetch behavior --- .../backend/community_interfaces_directory.py | 46 ++++++- meshchatx/src/backend/markdown_renderer.py | 7 +- meshchatx/src/backend/translator_handler.py | 55 ++++---- .../test_community_interfaces_directory.py | 117 ++++++++++++++++++ tests/backend/test_database.py | 73 ----------- tests/backend/test_database_pragma_safety.py | 89 +++++++++++++ tests/backend/test_emergency_mode.py | 8 -- tests/backend/test_markdown_renderer.py | 15 +++ tests/backend/test_startup.py | 1 - .../test_translator_handler_extended.py | 91 +++++++++++++- 10 files changed, 391 insertions(+), 111 deletions(-) create mode 100644 tests/backend/test_database_pragma_safety.py diff --git a/meshchatx/src/backend/community_interfaces_directory.py b/meshchatx/src/backend/community_interfaces_directory.py index f2cf940..9d8d09d 100644 --- a/meshchatx/src/backend/community_interfaces_directory.py +++ b/meshchatx/src/backend/community_interfaces_directory.py @@ -9,6 +9,7 @@ import re import urllib.error import urllib.request from typing import Any +from urllib.parse import urlparse DEFAULT_SUBMITTED_URL = ( "https://directory.rns.recipes/api/directory/submitted?search=&type=&status=online" @@ -16,6 +17,44 @@ DEFAULT_SUBMITTED_URL = ( DESCRIPTION = "directory.rns.recipes (user-submitted, online)" +_ALLOWED_DIRECTORY_HOSTS = frozenset({"directory.rns.recipes"}) + + +def validate_directory_fetch_url(url: str) -> str: + """Reject SSRF: only https to directory.rns.recipes, no credentials.""" + if not url or not isinstance(url, str): + msg = "URL must be a non-empty string" + raise ValueError(msg) + parsed = urlparse(url.strip()) + if parsed.scheme != "https": + msg = "Community directory URL must use https" + raise ValueError(msg) + netloc = parsed.netloc or "" + if "@" in netloc: + msg = "Community directory URL must not contain credentials" + raise ValueError(msg) + host = (parsed.hostname or "").lower() + if host not in _ALLOWED_DIRECTORY_HOSTS: + msg = "Community directory URL host is not allowed" + raise ValueError(msg) + return url.strip() + + +class _DirectoryFetchNoRedirectHandler(urllib.request.HTTPRedirectHandler): + """urllib follows redirects by default; blocked to prevent SSRF via Location.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + raise urllib.error.HTTPError( + req.full_url, + code, + "Redirects are not followed for community directory fetch", + headers, + fp, + ) + + +_DIRECTORY_FETCH_OPENER = urllib.request.build_opener(_DirectoryFetchNoRedirectHandler()) + def fetch_directory_payload(url: str, *, timeout: float = 60.0) -> object: req = urllib.request.Request( @@ -26,7 +65,7 @@ def fetch_directory_payload(url: str, *, timeout: float = 60.0) -> object: }, method="GET", ) - with urllib.request.urlopen(req, timeout=timeout) as resp: + with _DIRECTORY_FETCH_OPENER.open(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) @@ -35,7 +74,10 @@ def build_interfaces_from_directory_url( *, timeout: float = 60.0, ) -> tuple[list[dict[str, Any]], str]: - resolved = url or DEFAULT_SUBMITTED_URL + if url is not None and str(url).strip(): + resolved = validate_directory_fetch_url(url) + else: + resolved = DEFAULT_SUBMITTED_URL payload = fetch_directory_payload(resolved, timeout=timeout) rows = rows_from_payload(payload) return transform_directory_rows(rows), resolved diff --git a/meshchatx/src/backend/markdown_renderer.py b/meshchatx/src/backend/markdown_renderer.py index 55a48e1..5e792e8 100644 --- a/meshchatx/src/backend/markdown_renderer.py +++ b/meshchatx/src/backend/markdown_renderer.py @@ -10,9 +10,14 @@ _UNSAFE_PROTOCOLS = ("javascript:", "data:", "vbscript:", "file:") def _safe_href(url): if not url or not isinstance(url, str): return "#" - u = url.strip().lower() + trimmed = url.strip() + u = trimmed.lower() if any(u.startswith(p) for p in _UNSAFE_PROTOCOLS): return "#" + if u.startswith("//"): + return "#" + if trimmed.startswith("\\\\"): + return "#" if any(u.startswith(p) for p in _SAFE_LINK_PREFIXES): return url if ":" in u.split("/")[0]: diff --git a/meshchatx/src/backend/translator_handler.py b/meshchatx/src/backend/translator_handler.py index cb219ef..349f097 100644 --- a/meshchatx/src/backend/translator_handler.py +++ b/meshchatx/src/backend/translator_handler.py @@ -142,27 +142,35 @@ class TranslatorHandler: libretranslate_reachable = False url = libretranslate_url or self.libretranslate_url + explicit_override = ( + libretranslate_url is not None and str(libretranslate_url).strip() != "" + ) + libre_base = None if self.has_requests: - if libretranslate_url is not None and str(libretranslate_url).strip(): - try: - url = normalize_loopback_http_service_base(libretranslate_url) - except UnsafeOutboundUrlError as e: + try: + libre_base = normalize_loopback_http_service_base(url) + except UnsafeOutboundUrlError as e: + if explicit_override: msg = str(e) raise ValueError(msg) from e - try: - libretranslate_langs = _sync_run_coro(self._fetch_languages_async(url)) - if libretranslate_langs is not None: - libretranslate_reachable = True - languages.extend( - { - "code": lang.get("code"), - "name": lang.get("name"), - "source": "libretranslate", - } - for lang in libretranslate_langs + libre_base = None + if libre_base is not None: + try: + libretranslate_langs = _sync_run_coro( + self._fetch_languages_async(libre_base), ) - except Exception as e: - print(f"Failed to fetch LibreTranslate languages: {e}") + if libretranslate_langs is not None: + libretranslate_reachable = True + languages.extend( + { + "code": lang.get("code"), + "name": lang.get("name"), + "source": "libretranslate", + } + for lang in libretranslate_langs + ) + except Exception as e: + print(f"Failed to fetch LibreTranslate languages: {e}") if self.has_argos_lib: try: @@ -222,14 +230,13 @@ class TranslatorHandler: return self._translate_argos(text, source_lang, target_lang) if self.translator_libretranslate_enabled and self.has_requests: + url_raw = libretranslate_url or self.libretranslate_url + try: + url = normalize_loopback_http_service_base(url_raw) + except UnsafeOutboundUrlError as e: + msg = str(e) + raise ValueError(msg) from e try: - url = libretranslate_url or self.libretranslate_url - if libretranslate_url is not None and str(libretranslate_url).strip(): - try: - url = normalize_loopback_http_service_base(libretranslate_url) - except UnsafeOutboundUrlError as e: - msg = str(e) - raise ValueError(msg) from e return self._translate_libretranslate( text, source_lang=source_lang, diff --git a/tests/backend/test_community_interfaces_directory.py b/tests/backend/test_community_interfaces_directory.py index ea663f7..90b193f 100644 --- a/tests/backend/test_community_interfaces_directory.py +++ b/tests/backend/test_community_interfaces_directory.py @@ -1,13 +1,20 @@ # SPDX-License-Identifier: 0BSD +import threading +import urllib.error +import urllib.request +from http.server import BaseHTTPRequestHandler, HTTPServer + import pytest from hypothesis import given, settings from hypothesis import strategies as st from meshchatx.src.backend.community_interfaces_directory import ( DEFAULT_SUBMITTED_URL, + fetch_directory_payload, rows_from_payload, transform_directory_rows, + validate_directory_fetch_url, ) @@ -16,6 +23,116 @@ def test_default_url_is_submitted_online(): assert "status=online" in DEFAULT_SUBMITTED_URL +def test_validate_directory_fetch_url_accepts_default_host(): + u = "https://directory.rns.recipes/api/foo?bar=1" + assert validate_directory_fetch_url(u) == u + + +@pytest.mark.parametrize( + "bad", + [ + "http://directory.rns.recipes/api", + "https://127.0.0.1/", + "https://metadata.internal/", + "ftp://directory.rns.recipes/", + "https://evil.com/https://directory.rns.recipes/", + "https://user:pass@directory.rns.recipes/", + "https://not-directory.rns.recipes.example/api", + ], +) +def test_validate_directory_fetch_url_rejects_ssrf(bad): + with pytest.raises(ValueError): + validate_directory_fetch_url(bad) + + +class _Redirect302Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(302) + self.send_header("Location", "http://127.0.0.1:9/") + self.end_headers() + + def log_message(self, *args): + return + + +class _Json200Handler(BaseHTTPRequestHandler): + def do_GET(self): + body = b'{"data":[]}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + return + + +@pytest.fixture +def redirect_http_port(): + srv = HTTPServer(("127.0.0.1", 0), _Redirect302Handler) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + port = srv.server_address[1] + yield port + srv.shutdown() + + +@pytest.fixture +def json_http_port(): + srv = HTTPServer(("127.0.0.1", 0), _Json200Handler) + thread = threading.Thread(target=srv.serve_forever, daemon=True) + thread.start() + port = srv.server_address[1] + yield port + srv.shutdown() + + +def test_directory_fetch_opener_blocks_http_redirect(redirect_http_port): + from meshchatx.src.backend.community_interfaces_directory import ( + _DIRECTORY_FETCH_OPENER, + ) + + req = urllib.request.Request(f"http://127.0.0.1:{redirect_http_port}/") + with pytest.raises(urllib.error.HTTPError) as ei: + _DIRECTORY_FETCH_OPENER.open(req, timeout=3) + assert ei.value.code == 302 + + +def test_fetch_directory_payload_reads_json_when_no_redirect( + monkeypatch, + json_http_port, +): + import meshchatx.src.backend.community_interfaces_directory as cid + + monkeypatch.setattr( + cid, + "validate_directory_fetch_url", + lambda url: url, + ) + out = fetch_directory_payload( + f"http://127.0.0.1:{json_http_port}/x", + timeout=3, + ) + assert out == {"data": []} + + +def test_fetch_directory_payload_raises_on_redirect(monkeypatch, redirect_http_port): + import meshchatx.src.backend.community_interfaces_directory as cid + + monkeypatch.setattr( + cid, + "validate_directory_fetch_url", + lambda url: url, + ) + with pytest.raises(urllib.error.HTTPError) as ei: + fetch_directory_payload( + f"http://127.0.0.1:{redirect_http_port}/", + timeout=3, + ) + assert ei.value.code == 302 + + def test_rows_from_payload_dict_data(): rows = rows_from_payload({"data": [{"id": 1}]}) assert rows == [{"id": 1}] diff --git a/tests/backend/test_database.py b/tests/backend/test_database.py index 4d2ef53..31aa5ef 100644 --- a/tests/backend/test_database.py +++ b/tests/backend/test_database.py @@ -1,13 +1,11 @@ # SPDX-License-Identifier: 0BSD import os -import sqlite3 import tempfile import pytest from meshchatx.src.backend.database import Database -from meshchatx.src.backend.database.legacy_migrator import LegacyMigrator from meshchatx.src.backend.database.provider import DatabaseProvider from meshchatx.src.backend.database.schema import DatabaseSchema @@ -43,77 +41,6 @@ def test_database_initialization(temp_db): provider.close() -def test_legacy_migrator_detection(temp_db): - # Setup current DB - provider = DatabaseProvider(temp_db) - schema = DatabaseSchema(provider) - schema.initialize() - - # Setup a "legacy" DB in a temp directory - with tempfile.TemporaryDirectory() as legacy_dir: - identity_hash = "deadbeef" - legacy_identity_dir = os.path.join(legacy_dir, "identities", identity_hash) - os.makedirs(legacy_identity_dir) - legacy_db_path = os.path.join(legacy_identity_dir, "database.db") - - legacy_conn = sqlite3.connect(legacy_db_path) - legacy_conn.execute("CREATE TABLE config (key TEXT, value TEXT)") - legacy_conn.execute( - "INSERT INTO config (key, value) VALUES ('display_name', 'Legacy User')", - ) - legacy_conn.commit() - legacy_conn.close() - - migrator = LegacyMigrator(provider, legacy_dir, identity_hash) - assert migrator.get_legacy_db_path() == legacy_db_path - assert migrator.should_migrate() is True - - provider.close() - - -def test_legacy_migration_data(temp_db): - provider = DatabaseProvider(temp_db) - schema = DatabaseSchema(provider) - schema.initialize() - - with tempfile.TemporaryDirectory() as legacy_dir: - identity_hash = "deadbeef" - legacy_identity_dir = os.path.join(legacy_dir, "identities", identity_hash) - os.makedirs(legacy_identity_dir) - legacy_db_path = os.path.join(legacy_identity_dir, "database.db") - - # Create legacy DB with some data - legacy_conn = sqlite3.connect(legacy_db_path) - legacy_conn.execute( - "CREATE TABLE lxmf_messages (hash TEXT UNIQUE, content TEXT)", - ) - legacy_conn.execute( - "INSERT INTO lxmf_messages (hash, content) VALUES ('msg1', 'Hello Legacy')", - ) - legacy_conn.execute("CREATE TABLE config (key TEXT UNIQUE, value TEXT)") - legacy_conn.execute( - "INSERT INTO config (key, value) VALUES ('test_key', 'test_val')", - ) - legacy_conn.commit() - legacy_conn.close() - - migrator = LegacyMigrator(provider, legacy_dir, identity_hash) - assert migrator.migrate() is True - - # Verify data moved - msg_row = provider.fetchone( - "SELECT content FROM lxmf_messages WHERE hash = 'msg1'", - ) - assert msg_row["content"] == "Hello Legacy" - - config_row = provider.fetchone( - "SELECT value FROM config WHERE key = 'test_key'", - ) - assert config_row["value"] == "test_val" - - provider.close() - - def test_database_health_snapshot_free_space(temp_db): db = Database(temp_db) db.initialize() diff --git a/tests/backend/test_database_pragma_safety.py b/tests/backend/test_database_pragma_safety.py new file mode 100644 index 0000000..01bcc95 --- /dev/null +++ b/tests/backend/test_database_pragma_safety.py @@ -0,0 +1,89 @@ +# SPDX-License-Identifier: 0BSD + +import os +import tempfile + +import pytest + +from meshchatx.src.backend.database import ( + Database, + _sanitize_pragma_read_name, + _sanitize_wal_checkpoint_mode, +) +from meshchatx.src.backend.database.provider import DatabaseProvider + + +@pytest.fixture +def temp_db_path(): + fd, path = tempfile.mkstemp() + os.close(fd) + yield path + if os.path.exists(path): + os.remove(path) + wal = path + "-wal" + shm = path + "-shm" + for p in (wal, shm): + if os.path.exists(p): + os.remove(p) + + +@pytest.fixture(autouse=True) +def reset_db_singleton(temp_db_path): + DatabaseProvider._instance = None + yield + DatabaseProvider._instance = None + + +def test_sanitize_pragma_read_name_accepts_known_tokens(): + assert _sanitize_pragma_read_name("journal_mode") == "journal_mode" + assert _sanitize_pragma_read_name(" page_size ") == "page_size" + + +@pytest.mark.parametrize( + "bad", + [ + "", + " ", + "journal mode", + "journal-mode", + "x;detach", + "x'y", + "../../../x", + "pragma(x)", + ], +) +def test_sanitize_pragma_read_name_rejects_injection(bad): + assert _sanitize_pragma_read_name(bad) is None + + +def test_sanitize_wal_checkpoint_mode_accepts_keywords(): + assert _sanitize_wal_checkpoint_mode("truncate") == "TRUNCATE" + assert _sanitize_wal_checkpoint_mode("PASSIVE") == "PASSIVE" + + +def test_sanitize_wal_checkpoint_mode_rejects_injection(): + with pytest.raises(ValueError): + _sanitize_wal_checkpoint_mode("FULL);VACUUM") + with pytest.raises(ValueError): + _sanitize_wal_checkpoint_mode("bogus") + + +def test_get_pragma_value_returns_default_for_malicious_name(temp_db_path): + db = Database(temp_db_path) + db.initialize() + assert db._get_pragma_value("journal_mode;evil", "fallback") == "fallback" + jm = db._get_pragma_value("journal_mode") + assert jm is not None and str(jm).lower() == "wal" + + +def test_checkpoint_wal_rejects_injected_mode(temp_db_path): + db = Database(temp_db_path) + db.initialize() + with pytest.raises(ValueError): + db._checkpoint_wal("TRUNCATE);ATTACH DATABASE 'x' AS z") + + +def test_checkpoint_wal_passive_succeeds(temp_db_path): + db = Database(temp_db_path) + db.initialize() + db._checkpoint_wal("PASSIVE") diff --git a/tests/backend/test_emergency_mode.py b/tests/backend/test_emergency_mode.py index 7bfdcee..3a36ea5 100644 --- a/tests/backend/test_emergency_mode.py +++ b/tests/backend/test_emergency_mode.py @@ -113,10 +113,6 @@ def test_emergency_mode_startup_logic(mock_rns, temp_dir): mock_integrity_instance = mock_integrity_class.return_value assert mock_integrity_instance.check_integrity.call_count == 0 - # Verify migrate_from_legacy was NOT called - mock_db_instance = mock_db_class.return_value - assert mock_db_instance.migrate_from_legacy.call_count == 0 - # Verify TelephoneManager.init_telephone was NOT called mock_tel_instance = mock_tel_class.return_value assert mock_tel_instance.init_telephone.call_count == 0 @@ -215,10 +211,6 @@ def test_normal_mode_startup_logic(mock_rns, temp_dir): # Verify IntegrityManager.check_integrity WAS called assert mock_integrity_instance.check_integrity.call_count == 1 - # Verify migrate_from_legacy WAS called - mock_db_instance = mock_db_class.return_value - assert mock_db_instance.migrate_from_legacy.call_count == 1 - # Verify TelephoneManager.init_telephone WAS called mock_tel_instance = mock_tel_class.return_value assert mock_tel_instance.init_telephone.call_count == 1 diff --git a/tests/backend/test_markdown_renderer.py b/tests/backend/test_markdown_renderer.py index 8014d27..6229e0c 100644 --- a/tests/backend/test_markdown_renderer.py +++ b/tests/backend/test_markdown_renderer.py @@ -118,6 +118,21 @@ class TestMarkdownRenderer(unittest.TestCase): self.assertNotIn("vbscript:", r) self.assertIn('href="#"', r) + def test_protocol_relative_link_href_neutralized(self): + r = MarkdownRenderer.render("[phish](//evil.example/phish)") + self.assertNotIn("//evil.example", r) + self.assertNotIn('href="//', r) + self.assertIn('href="#"', r) + + def test_unc_link_href_neutralized(self): + md = "[click](" + "\\\\\\\\evil.example\\\\share" + ")" + r = MarkdownRenderer.render(md) + self.assertIn('href="#"', r) + + def test_protocol_relative_image_src_neutralized(self): + r = MarkdownRenderer.render("![x](//evil.example/i)") + self.assertNotIn("//evil.example", r) + def test_safe_links_preserved(self): r = MarkdownRenderer.render("[link](https://example.com/path)") self.assertIn('href="https://example.com/path"', r) diff --git a/tests/backend/test_startup.py b/tests/backend/test_startup.py index 699321c..3f56547 100644 --- a/tests/backend/test_startup.py +++ b/tests/backend/test_startup.py @@ -142,7 +142,6 @@ def test_reticulum_meshchat_init(mock_rns, temp_dir): # Verify database initialization mock_db_instance.initialize.assert_called_once() - mock_db_instance.migrate_from_legacy.assert_called_once() # Verify RNS initialization mock_rns["Reticulum"].assert_called_once_with(temp_dir) diff --git a/tests/backend/test_translator_handler_extended.py b/tests/backend/test_translator_handler_extended.py index d11ca76..82629a9 100644 --- a/tests/backend/test_translator_handler_extended.py +++ b/tests/backend/test_translator_handler_extended.py @@ -9,11 +9,11 @@ from meshchatx.src.backend.translator_handler import TranslatorHandler def test_translator_handler_init(): handler = TranslatorHandler( - libretranslate_url="http://test:5000", + libretranslate_url="http://127.0.0.1:5000", translator_argos_enabled=True, translator_libretranslate_enabled=True, ) - assert handler.libretranslate_url == "http://test:5000" + assert handler.libretranslate_url == "http://127.0.0.1:5000" assert handler.translator_argos_enabled is True @@ -130,6 +130,56 @@ def test_detect_language_libretranslate(mock_session_cls): assert result["source_lang"] == "en" +@patch("meshchatx.src.backend.translator_handler.aiohttp.ClientSession") +def test_libretranslate_post_disallows_redirects(mock_session_cls): + mock_response = MagicMock() + mock_response.status = 200 + mock_response.json = AsyncMock( + return_value={ + "translatedText": "y", + "detectedLanguage": {"language": "en"}, + }, + ) + mock_post_ctx = MagicMock() + mock_post_ctx.__aenter__ = AsyncMock(return_value=mock_response) + mock_post_ctx.__aexit__ = AsyncMock(return_value=None) + mock_session = MagicMock() + mock_session.post = MagicMock(return_value=mock_post_ctx) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + mock_session_cls.return_value = mock_session + + handler = TranslatorHandler(translator_libretranslate_enabled=True) + handler.has_requests = True + handler.translate_text("Hello", source_lang="en", target_lang="fr", use_argos=False) + + assert mock_session.post.call_args.kwargs.get("allow_redirects") is False + + +@patch("meshchatx.src.backend.translator_handler.aiohttp.ClientSession") +def test_libretranslate_get_languages_disallows_redirects(mock_session_cls): + mock_response = MagicMock() + mock_response.status = 200 + mock_response.json = AsyncMock(return_value=[{"code": "en", "name": "English"}]) + mock_get_ctx = MagicMock() + mock_get_ctx.__aenter__ = AsyncMock(return_value=mock_response) + mock_get_ctx.__aexit__ = AsyncMock(return_value=None) + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_get_ctx) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + mock_session_cls.return_value = mock_session + + handler = TranslatorHandler() + handler.has_argos = False + handler.has_argos_lib = False + handler.has_argos_cli = False + handler.has_requests = True + handler.get_supported_languages() + + assert mock_session.get.call_args.kwargs.get("allow_redirects") is False + + def test_translator_handler_errors(): handler = TranslatorHandler( translator_argos_enabled=False, @@ -149,3 +199,40 @@ def test_language_code_to_name(): assert LANGUAGE_CODE_TO_NAME["en"] == "English" assert LANGUAGE_CODE_TO_NAME["de"] == "German" + + +@patch("meshchatx.src.backend.translator_handler.aiohttp.ClientSession") +def test_get_supported_languages_skips_non_loopback_stored_url(mock_session_cls): + handler = TranslatorHandler( + libretranslate_url="http://169.254.169.254/", + translator_libretranslate_enabled=True, + ) + handler.has_requests = True + handler.has_argos = False + handler.has_argos_lib = False + handler.has_argos_cli = False + assert handler.get_supported_languages() == [] + mock_session_cls.assert_not_called() + + +def test_translate_text_rejects_non_loopback_libre_url(): + handler = TranslatorHandler( + libretranslate_url="http://example.com:5000/", + translator_libretranslate_enabled=True, + translator_argos_enabled=False, + ) + handler.has_requests = True + with pytest.raises(ValueError, match="URL host must be"): + handler.translate_text("Hello", "en", "de", use_argos=False) + + +def test_get_translator_languages_response_explicit_bad_override_raises(): + handler = TranslatorHandler( + libretranslate_url="http://127.0.0.1:5000", + translator_libretranslate_enabled=True, + ) + handler.has_requests = True + with pytest.raises(ValueError, match="URL host must be"): + handler.get_translator_languages_response( + libretranslate_url="http://metadata.example/latest", + )