You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by ji...@apache.org on 2026/10/10 09:24:51 UTC
(ambari-mpacks) 07/35: AMBARI-26663: Add verified runtime installation and isolated observers
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ambari-mpacks.git
commit b2ab654dff7efd832d2d380de49d3a7552e10587
Author: jialiang <ji...@gmail.com>
AuthorDate: Mon Sep 28 12:11:22 2026 +0800
AMBARI-26663: Add verified runtime installation and isolated observers
---
shared/managed_service_v2.py | 68 ++++++++++++++
shared/managed_topology.py | 33 +++++++
shared/verified_archive.py | 186 +++++++++++++++++++++++++++++++++++++++
tests/test_managed_service_v2.py | 65 ++++++++++++++
tests/test_verified_archive.py | 138 +++++++++++++++++++++++++++++
5 files changed, 490 insertions(+)
diff --git a/shared/managed_service_v2.py b/shared/managed_service_v2.py
new file mode 100644
index 0000000..71ecd3a
--- /dev/null
+++ b/shared/managed_service_v2.py
@@ -0,0 +1,68 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import json
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+
+from resource_management.core.exceptions import Fail
+from resource_management.libraries.script.script import Script
+from managed_runtime import validate_result
+from managed_service import ManagedService
+from managed_topology import assigned_hosts
+from observations import observer_python
+
+
+def observer_environment():
+ return {key: value for key, value in os.environ.items() if key not in {"PYTHONPATH", "PYTHONHOME"}}
+
+
+class ManagedServiceV2(ManagedService):
+ topology_fields = {}
+ install_timeout = 7200
+
+ def observe(self, action, publish=True):
+ config = Script.get_config()
+ command = config["commandParams"]
+ request = {
+ "schema_version": 1, "definition_snapshot": command["mpack_definition_snapshot"],
+ "execution_id": command["mpack_execution_id"], "task_id": config.get("taskId"),
+ "action": action,
+ "parameters": assigned_hosts(config["configurations"][self.service + "-env"],
+ config.get("clusterHostInfo", {}), self.topology_fields),
+ }
+ request["parameters"]["component"] = self.component
+ request["parameters"]["hostname"] = config["agentLevelParams"]["hostname"]
+ with tempfile.TemporaryDirectory(prefix="ambari-" + self.service + "-") as temporary:
+ os.chmod(temporary, 0o700)
+ source, destination = Path(temporary) / "request.json", Path(temporary) / "result.json"
+ source.write_text(json.dumps(request), encoding="utf-8")
+ process = subprocess.run([observer_python(), str(Path(__file__).with_name("runtime.py")),
+ str(source), str(destination)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+ timeout=self.install_timeout if action == "install" else 900, check=False,
+ env=observer_environment())
+ if not destination.is_file() or destination.stat().st_size > 1024 * 1024:
+ raise Fail("Service action has no bounded structured result")
+ try:
+ observation = validate_result(request, json.loads(destination.read_text()))
+ except (TypeError, ValueError):
+ raise Fail("Service result has no matching execution identity")
+ if publish:
+ self.put_structured_out({"software_observation": observation})
+ if process.returncode != 0 or observation["outcome"] != "APPLIED":
+ raise Fail("Service action is unresolved; inspect its structured error type")
+ return observation
diff --git a/shared/managed_topology.py b/shared/managed_topology.py
new file mode 100644
index 0000000..c1dc76d
--- /dev/null
+++ b/shared/managed_topology.py
@@ -0,0 +1,33 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from managed_runtime import host_value
+
+
+def assigned_hosts(parameters, topology, fields):
+ result = dict(parameters)
+ for field, (component_key, single, optional) in fields.items():
+ assigned = topology.get(component_key, [] if optional else None)
+ if not isinstance(assigned, (list, tuple)) or any(not isinstance(host, str) for host in assigned):
+ raise ValueError("Missing component assignment: " + component_key)
+ hosts = sorted(host_value(host) for host in assigned)
+ if len(hosts) != len(set(hosts)) or (not hosts and not optional) or (single and len(hosts) != 1):
+ raise ValueError("Invalid component assignment: " + component_key)
+ configured = result.get(field)
+ selected = hosts[0] if single else ",".join(hosts)
+ if configured and configured != selected:
+ raise ValueError("Configured topology differs from assigned hosts: " + field)
+ result[field] = selected
+ return result
diff --git a/shared/verified_archive.py b/shared/verified_archive.py
new file mode 100644
index 0000000..e9a8574
--- /dev/null
+++ b/shared/verified_archive.py
@@ -0,0 +1,186 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import fcntl
+import hashlib
+import json
+import os
+import platform
+import re
+import shutil
+import tarfile
+import tempfile
+from pathlib import Path
+
+from managed_runtime import directory, download, no_links
+from observations import atomic_json
+
+ROOT = Path("/opt/ambari-mpacks")
+CACHE = Path("/var/cache/ambari-mpacks")
+
+
+def require_platform():
+ if platform.system() != "Linux" or platform.machine() not in {"aarch64", "arm64"}:
+ raise ValueError("This distribution requires Linux ARM64")
+
+
+def digest_file(path, algorithm):
+ digest = hashlib.new(algorithm)
+ with Path(path).open("rb") as stream:
+ for block in iter(lambda: stream.read(1024 * 1024), b""):
+ digest.update(block)
+ return digest.hexdigest()
+
+
+def _inside(root, path):
+ try:
+ path.relative_to(root)
+ except ValueError:
+ raise ValueError("Archive member leaves its distribution")
+
+
+def extract(archive, target, prefix, maximum_expanded=16 * 1024 ** 3):
+ target = Path(target)
+ no_links(target)
+ target.mkdir(mode=0o755)
+ pending_links = []
+ seen = set()
+ expanded = 0
+ with tarfile.open(str(archive), "r:gz") as source:
+ for entry in source:
+ parts = Path(entry.name).parts
+ if not parts or parts[0] != prefix or ".." in parts or Path(entry.name).is_absolute():
+ raise ValueError("Archive member leaves its pinned distribution")
+ relative = Path(*parts[1:])
+ if relative == Path("."):
+ if not entry.isdir() or entry.size:
+ raise ValueError("Invalid distribution root")
+ continue
+ if len(seen) >= 250000 or str(relative) in seen:
+ raise ValueError("Duplicate or excessive archive members")
+ seen.add(str(relative))
+ destination = target / relative
+ no_links(destination)
+ if entry.isdir() and entry.size == 0:
+ destination.mkdir(parents=True, exist_ok=True)
+ destination.chmod(0o755)
+ elif entry.isfile() and not entry.issparse():
+ expanded += entry.size
+ if expanded > maximum_expanded:
+ raise ValueError("Expanded distribution exceeds its limit")
+ destination.parent.mkdir(parents=True, exist_ok=True)
+ with source.extractfile(entry) as incoming, destination.open("xb") as outgoing:
+ shutil.copyfileobj(incoming, outgoing)
+ destination.chmod(0o755 if entry.mode & 0o111 else 0o644)
+ elif (entry.issym() or entry.islnk()) and entry.size == 0:
+ pending_links.append((destination, entry.linkname, entry.issym()))
+ else:
+ raise ValueError("Unsupported distribution member")
+ for destination, link, symbolic in pending_links:
+ link_path = Path(link)
+ if link_path.is_absolute() or ".." in link_path.parts and not symbolic:
+ raise ValueError("Invalid distribution link")
+ candidate = destination.parent / link if symbolic else target / link_path.relative_to(prefix)
+ resolved = candidate.resolve()
+ _inside(target.resolve(), resolved)
+ if not resolved.is_file() and not resolved.is_dir():
+ raise ValueError("Distribution link target is missing")
+ destination.parent.mkdir(parents=True, exist_ok=True)
+ if symbolic:
+ destination.symlink_to(link)
+ elif resolved.is_file():
+ os.link(str(resolved), str(destination))
+ else:
+ raise ValueError("Hard link does not point to a file")
+ return len(seen)
+
+
+def verify_install(home, name, version, archive_sha512, required):
+ home = Path(home)
+ no_links(home)
+ marker = home / ".ambari-receipt.json"
+ no_links(marker)
+ receipt = json.loads(marker.read_text())
+ identity = {"schema_version": 1, "name": name, "version": version,
+ "archive_sha512": archive_sha512}
+ if any(type(receipt.get(key)) is not type(value) or receipt[key] != value
+ for key, value in identity.items()):
+ raise ValueError("Installed distribution identity differs")
+ hashes = receipt.get("required_sha256")
+ if not isinstance(hashes, dict) or set(hashes) != set(required):
+ raise ValueError("Missing required distribution identity")
+ for relative in required:
+ path = home / relative
+ no_links(path)
+ if not path.is_file() or not isinstance(hashes[relative], str) or not re.fullmatch(r"[0-9a-f]{64}", hashes[relative]):
+ raise ValueError("Required distribution file is missing")
+ if digest_file(path, "sha256") != hashes[relative]:
+ raise ValueError("Installed distribution file differs from its receipt")
+ return {"name": name, "version": version, "archive_sha512": archive_sha512,
+ "verified_files": len(required)}
+
+
+def install_archive(name, version, url, archive_sha512, prefix, required,
+ maximum_archive=6 * 1024 ** 3, maximum_expanded=16 * 1024 ** 3):
+ require_platform()
+ if not re.fullmatch(r"[a-z][a-z0-9-]*", name) or not re.fullmatch(r"[A-Za-z0-9._-]+", version):
+ raise ValueError("Invalid distribution identity")
+ if not re.fullmatch(r"[0-9a-f]{128}", archive_sha512):
+ raise ValueError("Invalid pinned archive digest")
+ if not required or any(Path(path).is_absolute() or ".." in Path(path).parts
+ or Path(path) == Path(".") for path in required):
+ raise ValueError("Invalid required file manifest")
+ home = ROOT / name / version
+ directory(home.parent, "root", 0o755)
+ lock = home.parent / ".install.lock"
+ no_links(lock)
+ descriptor = os.open(str(lock), os.O_WRONLY | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600)
+ try:
+ fcntl.flock(descriptor, fcntl.LOCK_EX)
+ if home.exists():
+ return verify_install(home, name, version, archive_sha512, required)
+ return _install_locked(home, name, version, url, archive_sha512, prefix,
+ required, maximum_archive, maximum_expanded)
+ finally:
+ fcntl.flock(descriptor, fcntl.LOCK_UN)
+ os.close(descriptor)
+
+
+def _install_locked(home, name, version, url, archive_sha512, prefix, required,
+ maximum_archive, maximum_expanded):
+ cache = directory(CACHE, "root", 0o755) / (name + "-" + version + ".tar.gz")
+ no_links(cache)
+ with tempfile.TemporaryDirectory(prefix=".install-", dir=str(home.parent)) as work:
+ archive = cache
+ if cache.exists():
+ if digest_file(cache, "sha512") != archive_sha512:
+ raise ValueError("Cached distribution differs from its pinned digest")
+ else:
+ archive = Path(work) / "distribution.tar.gz"
+ download(url, archive, archive_sha512, maximum=maximum_archive)
+ target = Path(work) / "distribution"
+ extract(archive, target, prefix, maximum_expanded)
+ hashes = {}
+ for relative in required:
+ path = target / relative
+ no_links(path)
+ if not path.is_file():
+ raise ValueError("Pinned distribution lacks a required file")
+ hashes[relative] = digest_file(path, "sha256")
+ atomic_json(target / ".ambari-receipt.json", {"schema_version": 1, "name": name,
+ "version": version, "archive_sha512": archive_sha512,
+ "required_sha256": hashes})
+ os.rename(str(target), str(home))
+ return verify_install(home, name, version, archive_sha512, required)
diff --git a/tests/test_managed_service_v2.py b/tests/test_managed_service_v2.py
new file mode 100644
index 0000000..2cd670f
--- /dev/null
+++ b/tests/test_managed_service_v2.py
@@ -0,0 +1,65 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import importlib.util
+import json
+import os
+import sys
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "shared"))
+
+
+class ManagedServiceV2Tests(unittest.TestCase):
+ def test_observer_gets_clean_python_environment_and_matching_result(self):
+ config = {"commandParams": {"mpack_definition_snapshot": "a" * 64,
+ "mpack_execution_id": "12345678-1234-4234-8234-123456789abc"},
+ "taskId": 12, "configurations": {"test-env": {}},
+ "agentLevelParams": {"hostname": "node.test"}, "clusterHostInfo": {}}
+ spec = importlib.util.spec_from_file_location("managed_service_v2_test", str(ROOT / "shared/managed_service_v2.py"))
+ module = importlib.util.module_from_spec(spec)
+ with patch.dict(sys.modules, {
+ "resource_management.core.exceptions": SimpleNamespace(Fail=RuntimeError),
+ "resource_management.libraries.script.script": SimpleNamespace(Script=SimpleNamespace(get_config=lambda: config)),
+ "managed_service": SimpleNamespace(ManagedService=object),
+ }):
+ spec.loader.exec_module(module)
+ observed = []
+ def run(command, **kwargs):
+ self.assertNotIn("PYTHONPATH", kwargs["env"])
+ self.assertNotIn("PYTHONHOME", kwargs["env"])
+ self.assertEqual(kwargs["env"]["HTTPS_PROXY"], "http://proxy.test:7890")
+ request = json.loads(Path(command[-2]).read_text())
+ result = {key: request[key] for key in ("schema_version", "definition_snapshot", "execution_id", "task_id", "action")}
+ result.update(outcome="APPLIED", observations={"unit": "active"})
+ Path(command[-1]).write_text(json.dumps(result))
+ return SimpleNamespace(returncode=0)
+ instance = module.ManagedServiceV2()
+ instance.service, instance.component = "test", "node"
+ instance.put_structured_out = observed.append
+ with patch.dict(os.environ, {"PYTHONPATH": "/agent/python39", "PYTHONHOME": "/agent",
+ "HTTPS_PROXY": "http://proxy.test:7890"}), patch.object(
+ module, "observer_python", return_value="/usr/libexec/platform-python"), patch.object(
+ module.subprocess, "run", side_effect=run):
+ self.assertEqual(instance.observe("status")["task_id"], 12)
+ self.assertEqual(observed[0]["software_observation"]["outcome"], "APPLIED")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_verified_archive.py b/tests/test_verified_archive.py
new file mode 100644
index 0000000..98d89ac
--- /dev/null
+++ b/tests/test_verified_archive.py
@@ -0,0 +1,138 @@
+"""
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import hashlib
+import io
+import json
+import sys
+import tarfile
+import tempfile
+import time
+import unittest
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "shared"))
+from managed_topology import assigned_hosts
+import verified_archive
+from verified_archive import extract, verify_install
+
+
+def archive(path, entries):
+ with tarfile.open(str(path), "w:gz") as stream:
+ root = tarfile.TarInfo("software-1/")
+ root.type = tarfile.DIRTYPE
+ stream.addfile(root)
+ for name, kind, content in entries:
+ entry = tarfile.TarInfo(name)
+ entry.type = kind
+ if kind == tarfile.REGTYPE:
+ entry.size = len(content)
+ stream.addfile(entry, io.BytesIO(content))
+ else:
+ entry.linkname = content
+ stream.addfile(entry)
+
+
+class VerifiedArchiveTests(unittest.TestCase):
+ def test_extracts_only_contained_members_and_verifies_receipt(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary).resolve()
+ source = root / "software.tar.gz"
+ archive(source, [("software-1/bin/tool", tarfile.REGTYPE, b"binary"),
+ ("software-1/link", tarfile.SYMTYPE, "bin/tool")])
+ target = root / "installed"
+ self.assertEqual(extract(source, target, "software-1"), 2)
+ self.assertEqual((target / "link").read_bytes(), b"binary")
+ sha = hashlib.sha256(b"binary").hexdigest()
+ (target / ".ambari-receipt.json").write_text(json.dumps({
+ "schema_version": 1, "name": "software", "version": "1",
+ "archive_sha512": "a" * 128, "required_sha256": {"bin/tool": sha}}))
+ self.assertEqual(verify_install(target, "software", "1", "a" * 128,
+ ["bin/tool"])["verified_files"], 1)
+ (target / "bin/tool").write_bytes(b"foreign")
+ with self.assertRaises(ValueError):
+ verify_install(target, "software", "1", "a" * 128, ["bin/tool"])
+
+ def test_rejects_escape_duplicate_special_and_oversized_members(self):
+ cases = [
+ [("software-1/../escape", tarfile.REGTYPE, b"bad")],
+ [("software-1/file", tarfile.REGTYPE, b"one"),
+ ("software-1/file", tarfile.REGTYPE, b"two")],
+ [("software-1/link", tarfile.SYMTYPE, "../../outside")],
+ [("software-1/pipe", tarfile.FIFOTYPE, "")],
+ ]
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary).resolve()
+ for index, entries in enumerate(cases):
+ source = root / (str(index) + ".tar.gz")
+ archive(source, entries)
+ with self.subTest(index=index), self.assertRaises(ValueError):
+ extract(source, root / ("target-" + str(index)), "software-1")
+ source = root / "large.tar.gz"
+ archive(source, [("software-1/file", tarfile.REGTYPE, b"large")])
+ with self.assertRaises(ValueError):
+ extract(source, root / "large", "software-1", maximum_expanded=4)
+ self.assertFalse((root / "escape").exists())
+
+ def test_topology_rejects_foreign_missing_and_duplicate_hosts(self):
+ fields = {"coordinator_host": ("trino_coordinator_hosts", True, False),
+ "worker_hosts": ("trino_worker_hosts", False, True)}
+ topology = {"trino_coordinator_hosts": ["coordinator.test"],
+ "trino_worker_hosts": ["worker.test"]}
+ self.assertEqual(assigned_hosts({}, topology, fields),
+ {"coordinator_host": "coordinator.test", "worker_hosts": "worker.test"})
+ for parameters, assigned in [
+ ({"coordinator_host": "foreign.test"}, topology),
+ ({}, {"trino_coordinator_hosts": []}),
+ ({}, {"trino_coordinator_hosts": ["one.test", "two.test"]}),
+ ({}, {"trino_coordinator_hosts": ["one.test"],
+ "trino_worker_hosts": ["worker.test", "worker.test"]}),
+ ]:
+ with self.subTest(parameters=parameters, assigned=assigned), self.assertRaises(ValueError):
+ assigned_hosts(parameters, assigned, fields)
+
+ def test_parallel_installers_publish_one_verified_distribution(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary).resolve()
+ calls = []
+ def directory(path, *_):
+ Path(path).mkdir(parents=True, exist_ok=True)
+ return Path(path)
+ def install(home, *_):
+ calls.append(home)
+ time.sleep(0.05)
+ home.mkdir()
+ return {"version": "1"}
+ with patch.object(verified_archive, "require_platform"), patch.object(verified_archive, "ROOT", root), patch.object(
+ verified_archive, "directory", side_effect=directory), patch.object(
+ verified_archive, "_install_locked", side_effect=install), patch.object(
+ verified_archive, "verify_install", return_value={"version": "1"}):
+ with ThreadPoolExecutor(max_workers=2) as pool:
+ futures = [pool.submit(verified_archive.install_archive, "software", "1",
+ "https://example.test/software.tar.gz", "a" * 128, "software-1", ["bin/tool"])
+ for _ in range(2)]
+ self.assertEqual([future.result() for future in futures],
+ [{"version": "1"}, {"version": "1"}])
+ self.assertEqual(len(calls), 1)
+
+ def test_non_arm_linux_host_is_rejected_before_install(self):
+ with patch.object(verified_archive.platform, "system", return_value="Darwin"), self.assertRaises(ValueError):
+ verified_archive.require_platform()
+ with patch.object(verified_archive.platform, "system", return_value="Linux"), patch.object(
+ verified_archive.platform, "machine", return_value="x86_64"), self.assertRaises(ValueError):
+ verified_archive.require_platform()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ambari.apache.org
For additional commands, e-mail: commits-help@ambari.apache.org