21697093787

Committed 04 Feb 2026 09:56PM UTC coverage: 86.962% (-0.004%) from 86.966%

Build # 21697093787

Build Type

push

github

Committed by

web-flow

Commit Message

improve system information sent in session and container_info (#13680)

Run Details

10 of 17 new or added lines in 2 files covered. (58.82%)

222 existing lines in 17 files now uncovered.

70560 of 81139 relevant lines covered (86.96%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.21

/localstack-core/localstack/utils/analytics/metadata.py

import dataclasses
import logging
import os
import platform

from localstack import config
from localstack.constants import VERSION
from localstack.runtime import get_current_runtime, hooks
from localstack.utils.bootstrap import Container
from localstack.utils.files import rm_rf
from localstack.utils.functions import call_safe
from localstack.utils.json import FileMappedDocument
from localstack.utils.objects import singleton_factory
from localstack.utils.strings import long_uid, md5

LOG = logging.getLogger(__name__)

_PHYSICAL_ID_SALT = "ls"


@dataclasses.dataclass
class ClientMetadata:
    session_id: str
    machine_id: str
    api_key: str
    system: str
    version: str
    is_ci: bool
    is_docker: bool
    is_testing: bool
    product: str
    edition: str

    def __repr__(self):
        d = dataclasses.asdict(self)

        # anonymize api_key
        k = d.get("api_key")
        if k:
            k = "*" * len(k)
        d["api_key"] = k

        return f"ClientMetadata({d})"


def get_version_string() -> str:
    gh = config.LOCALSTACK_BUILD_GIT_HASH
    if gh:
        return f"{VERSION}:{gh}"
    else:
        return VERSION


def read_client_metadata() -> ClientMetadata:
    return ClientMetadata(
        session_id=get_session_id(),
        machine_id=get_machine_id(),
        api_key=get_api_key_or_auth_token() or "",  # api key should not be None
        system=get_system_information_summary(),
        version=get_version_string(),
        is_ci=os.getenv("CI") is not None,
        is_docker=config.is_in_docker,
        is_testing=config.is_local_test_mode(),
        product=get_localstack_product(),
        edition=os.getenv("LOCALSTACK_TELEMETRY_EDITION") or get_localstack_edition(),
    )


@singleton_factory
def get_session_id() -> str:
    """
    Returns the unique ID for this LocalStack session.
    :return: a UUID
    """
    return _generate_session_id()


@singleton_factory
def get_client_metadata() -> ClientMetadata:
    metadata = read_client_metadata()

    if config.DEBUG_ANALYTICS:
        LOG.info("resolved client metadata: %s", metadata)

    return metadata


@singleton_factory
def get_machine_id() -> str:
    cache_path = os.path.join(config.dirs.cache, "machine.json")
    try:
        doc = FileMappedDocument(cache_path)
    except Exception:
        # it's possible that the file is somehow messed up, so we try to delete the file first and try again.
        # if that fails, we return a generated ID.
        call_safe(rm_rf, args=(cache_path,))

        try:
            doc = FileMappedDocument(cache_path)
        except Exception:
            return _generate_machine_id()

    if "machine_id" not in doc:
        # generate a machine id
        doc["machine_id"] = _generate_machine_id()
        # try to cache the machine ID
        call_safe(doc.save)

    return doc["machine_id"]


def get_localstack_edition() -> str:
    # Generator expression to find the first hidden file ending with '-version'
    version_file = next(
        (
            f
            for f in os.listdir(config.dirs.static_libs)
            if f.startswith(".") and f.endswith("-version")
        ),
        None,
    )

    # Return the base name of the version file, or unknown if no file is found
    return version_file.removesuffix("-version").removeprefix(".") if version_file else "unknown"


def get_localstack_product() -> str:
    """
    Returns the telemetry product name from the env var, runtime, or "unknown".
    """
    try:
        runtime_product = get_current_runtime().components.name
    except ValueError:
        runtime_product = None

    return os.getenv("LOCALSTACK_TELEMETRY_PRODUCT") or runtime_product or "unknown"


def is_license_activated() -> bool:
    try:
        from localstack.pro.core import config  # noqa
    except ImportError:
        return False

    try:
        from localstack.pro.core.bootstrap import licensingv2

        return licensingv2.get_licensed_environment().activated
    except Exception:
        LOG.error(
            "Could not determine license activation status",
            exc_info=LOG.isEnabledFor(logging.DEBUG),
        )
        return False


def _generate_session_id() -> str:
    return long_uid()


def _anonymize_physical_id(physical_id: str) -> str:
    """
    Returns 12 digits of the salted hash of the given physical ID.

    :param physical_id: the physical id
    :return: an anonymized 12 digit value representing the physical ID.
    """
    hashed = md5(_PHYSICAL_ID_SALT + physical_id)
    return hashed[:12]


def _generate_machine_id() -> str:
    try:
        # try to get a robust ID from the docker socket (which will be the same from the host and the
        # container)
        from localstack.utils.docker_utils import DOCKER_CLIENT

        docker_id = DOCKER_CLIENT.get_system_id()
        # some systems like podman don't return a stable ID, so we double-check that here
        if docker_id == DOCKER_CLIENT.get_system_id():
            return f"dkr_{_anonymize_physical_id(docker_id)}"
    except Exception:
        pass

    if config.is_in_docker:
        return f"gen_{long_uid()[:12]}"

    # this can potentially be useful when generated on the host using the CLI and then mounted into the
    # container via machine.json
    try:
        if os.path.exists("/etc/machine-id"):
            with open("/etc/machine-id") as fd:
                machine_id = str(fd.read()).strip()
                if machine_id:
                    return f"sys_{_anonymize_physical_id(machine_id)}"
    except Exception:
        pass

    # always fall back to a generated id
    return f"gen_{long_uid()[:12]}"


def get_api_key_or_auth_token() -> str | None:
    # TODO: this is duplicated code from ext, but should probably migrate that to localstack
    auth_token = os.environ.get("LOCALSTACK_AUTH_TOKEN", "").strip("'\" ")
    if auth_token:
        return auth_token

    api_key = os.environ.get("LOCALSTACK_API_KEY", "").strip("'\" ")
    if api_key:
        return api_key

    return None


@singleton_factory
def get_system() -> str:
    # TODO: candidate for removal
    try:
        # try to get the system from the docker socket
        from localstack.utils.docker_utils import DOCKER_CLIENT

        system = DOCKER_CLIENT.get_system_info()
        if system.get("OSType"):
            return system.get("OSType").lower()
    except Exception:
        pass

    if config.is_in_docker:
        return "docker"

    return platform.system().lower()


@singleton_factory
def get_system_information_summary() -> str:
    """
    Returns a string that contains three comma-separated values: The operating system, kernel version,
    and architecture. We either use the docker socket to resolve the information, if that is not available
    we fall back ``platform.uname()``. If we're in docker and we don't have the docker socket available,
    we add ``(Container)`` to the operating system type to indicate that we don't have any additional
    information.

    Some examples:

    If the Docker socket is available:
     - Docker Desktop,5.15.90.1-microsoft-standard-WSL2,x86_64
     - Linux Mint 21.1,5.19.0-32-generic,x86_64

    If the Docker socket is not available, and we're on the host:
     - Windows,10,AMD64
     - Linux,5.19.0-32-generic,x86_64

    If the Docker socket is not available, and we're in the container:
     - Linux(Container),5.19.0-32-generic,x86_64

    :return: A string representing the system's information
    """
    try:
        # try to get the system from the docker socket
        from localstack.utils.docker_utils import DOCKER_CLIENT

        system = DOCKER_CLIENT.get_system_info()

        return ",".join(
            [
                system["OperatingSystem"],
                system["KernelVersion"],
                system["Architecture"],
            ]
        )
    except Exception:
        if config.DEBUG_ANALYTICS:
            LOG.exception(
                "Unable to get system information from docker socket, falling back to platform.uname()"
            )

    uname = platform.uname()

    if config.is_in_docker:
        return ",".join(
            [
                f"{uname.system}(Container)",
                uname.release,
                uname.machine,
            ]
        )

    return ",".join(
        [
            uname.system,
            uname.release,
            uname.machine,
        ]
    )


@hooks.prepare_host()
def prepare_host_machine_id():
    # lazy-init machine ID into cache on the host, which can then be used in the container
    get_machine_id()


@hooks.configure_localstack_container()
def _mount_machine_file(container: Container):
    from localstack.utils.container_utils.container_client import BindMount

    # mount tha machine file from the host's CLI cache directory into the appropriate location in the
    # container
    machine_file = os.path.join(config.dirs.cache, "machine.json")
    if os.path.isfile(machine_file):
        target = os.path.join(config.dirs.for_container().cache, "machine.json")
        container.config.volumes.add(BindMount(machine_file, target, read_only=True))

1	import dataclasses	1✔
2	import logging	1✔
3	import os	1✔
4	import platform	1✔
5
6	from localstack import config	1✔
7	from localstack.constants import VERSION	1✔
8	from localstack.runtime import get_current_runtime, hooks	1✔
9	from localstack.utils.bootstrap import Container	1✔
10	from localstack.utils.files import rm_rf	1✔
11	from localstack.utils.functions import call_safe	1✔
12	from localstack.utils.json import FileMappedDocument	1✔
13	from localstack.utils.objects import singleton_factory	1✔
14	from localstack.utils.strings import long_uid, md5	1✔
15
16	LOG = logging.getLogger(__name__)	1✔
17
18	_PHYSICAL_ID_SALT = "ls"	1✔
19
20
21	@dataclasses.dataclass	1✔
22	class ClientMetadata:	1✔
23	session_id: str	1✔
24	machine_id: str	1✔
25	api_key: str	1✔
26	system: str	1✔
27	version: str	1✔
28	is_ci: bool	1✔
29	is_docker: bool	1✔
30	is_testing: bool	1✔
31	product: str	1✔
32	edition: str	1✔
33
34	def __repr__(self):
35	d = dataclasses.asdict(self)
36
37	# anonymize api_key
38	k = d.get("api_key")
39	if k:
40	k = "" len(k)
41	d["api_key"] = k
42
43	return f"ClientMetadata({d})"
44
45
46	def get_version_string() -> str:	1✔
47	gh = config.LOCALSTACK_BUILD_GIT_HASH	1✔
48	if gh:	1✔
49	return f"{VERSION}:{gh}"	1✔
50	else:
51	return VERSION	1✔
52
53
54	def read_client_metadata() -> ClientMetadata:	1✔
55	return ClientMetadata(	1✔
56	session_id=get_session_id(),
57	machine_id=get_machine_id(),
58	api_key=get_api_key_or_auth_token() or "", # api key should not be None
59	system=get_system_information_summary(),
60	version=get_version_string(),
61	is_ci=os.getenv("CI") is not None,
62	is_docker=config.is_in_docker,
63	is_testing=config.is_local_test_mode(),
64	product=get_localstack_product(),
65	edition=os.getenv("LOCALSTACK_TELEMETRY_EDITION") or get_localstack_edition(),
66	)
67
68
69	@singleton_factory	1✔
70	def get_session_id() -> str:	1✔
71	"""
72	Returns the unique ID for this LocalStack session.
73	:return: a UUID
74	"""
75	return _generate_session_id()	1✔
76
77
78	@singleton_factory	1✔
79	def get_client_metadata() -> ClientMetadata:	1✔
80	metadata = read_client_metadata()	1✔
81
82	if config.DEBUG_ANALYTICS:	1✔
83	LOG.info("resolved client metadata: %s", metadata)	×
84
85	return metadata	1✔
86
87
88	@singleton_factory	1✔
89	def get_machine_id() -> str:	1✔
90	cache_path = os.path.join(config.dirs.cache, "machine.json")	1✔
91	try:	1✔
92	doc = FileMappedDocument(cache_path)	1✔
93	except Exception:	×
94	# it's possible that the file is somehow messed up, so we try to delete the file first and try again.
95	# if that fails, we return a generated ID.
96	call_safe(rm_rf, args=(cache_path,))	×
97
98	try:	×
99	doc = FileMappedDocument(cache_path)	×
100	except Exception:	×
101	return _generate_machine_id()	×
102
103	if "machine_id" not in doc:	1✔
104	# generate a machine id
105	doc["machine_id"] = _generate_machine_id()	1✔
106	# try to cache the machine ID
107	call_safe(doc.save)	1✔
108
109	return doc["machine_id"]	1✔
110
111
112	def get_localstack_edition() -> str:	1✔
113	# Generator expression to find the first hidden file ending with '-version'
114	version_file = next(	1✔
115	(
116	f
117	for f in os.listdir(config.dirs.static_libs)
118	if f.startswith(".") and f.endswith("-version")
119	),
120	None,
121	)
122
123	# Return the base name of the version file, or unknown if no file is found
124	return version_file.removesuffix("-version").removeprefix(".") if version_file else "unknown"	1✔
125
126
127	def get_localstack_product() -> str:	1✔
128	"""
129	Returns the telemetry product name from the env var, runtime, or "unknown".
130	"""
131	try:	1✔
132	runtime_product = get_current_runtime().components.name	1✔
133	except ValueError:	×
134	runtime_product = None	×
135
136	return os.getenv("LOCALSTACK_TELEMETRY_PRODUCT") or runtime_product or "unknown"	1✔
137
138
139	def is_license_activated() -> bool:	1✔
140	try:	1✔
141	from localstack.pro.core import config # noqa	1✔
142	except ImportError:	1✔
143	return False	1✔
144
145	try:	×
146	from localstack.pro.core.bootstrap import licensingv2	×
147
148	return licensingv2.get_licensed_environment().activated	×
149	except Exception:	×
150	LOG.error(	×
151	"Could not determine license activation status",
152	exc_info=LOG.isEnabledFor(logging.DEBUG),
153	)
154	return False	×
155
156
157	def _generate_session_id() -> str:	1✔
158	return long_uid()	1✔
159
160
161	def _anonymize_physical_id(physical_id: str) -> str:	1✔
162	"""
163	Returns 12 digits of the salted hash of the given physical ID.
164
165	:param physical_id: the physical id
166	:return: an anonymized 12 digit value representing the physical ID.
167	"""
168	hashed = md5(_PHYSICAL_ID_SALT + physical_id)	1✔
169	return hashed[:12]	1✔
170
171
172	def _generate_machine_id() -> str:	1✔
173	try:	1✔
174	# try to get a robust ID from the docker socket (which will be the same from the host and the
175	# container)
176	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
177
178	docker_id = DOCKER_CLIENT.get_system_id()	1✔
179	# some systems like podman don't return a stable ID, so we double-check that here
180	if docker_id == DOCKER_CLIENT.get_system_id():	1✔
181	return f"dkr_{_anonymize_physical_id(docker_id)}"	1✔
182	except Exception:	×
183	pass	×
184
185	if config.is_in_docker:	×
186	return f"gen_{long_uid()[:12]}"	×
187
188	# this can potentially be useful when generated on the host using the CLI and then mounted into the
189	# container via machine.json
190	try:	×
191	if os.path.exists("/etc/machine-id"):	×
192	with open("/etc/machine-id") as fd:	×
193	machine_id = str(fd.read()).strip()	×
194	if machine_id:	×
195	return f"sys_{_anonymize_physical_id(machine_id)}"	×
196	except Exception:	×
197	pass	×
198
199	# always fall back to a generated id
200	return f"gen_{long_uid()[:12]}"	×
201
202
203	def get_api_key_or_auth_token() -> str \| None:	1✔
204	# TODO: this is duplicated code from ext, but should probably migrate that to localstack
205	auth_token = os.environ.get("LOCALSTACK_AUTH_TOKEN", "").strip("'\" ")	1✔
206	if auth_token:	1✔
207	return auth_token	×
208
209	api_key = os.environ.get("LOCALSTACK_API_KEY", "").strip("'\" ")	1✔
210	if api_key:	1✔
211	return api_key	×
212
213	return None	1✔
214
215
216	@singleton_factory	1✔
217	def get_system() -> str:	1✔
218	# TODO: candidate for removal
UNCOV 219	try:	×
220	# try to get the system from the docker socket
UNCOV 221	from localstack.utils.docker_utils import DOCKER_CLIENT	×
222
UNCOV 223	system = DOCKER_CLIENT.get_system_info()	×
UNCOV 224	if system.get("OSType"):	×
UNCOV 225	return system.get("OSType").lower()	×
226	except Exception:	×
227	pass	×
228
229	if config.is_in_docker:	×
230	return "docker"	×
231
232	return platform.system().lower()	×
233
234
235	@singleton_factory	1✔
236	def get_system_information_summary() -> str:	1✔
237	"""
238	Returns a string that contains three comma-separated values: The operating system, kernel version,
239	and architecture. We either use the docker socket to resolve the information, if that is not available
240	we fall back ``platform.uname()``. If we're in docker and we don't have the docker socket available,
241	we add ``(Container)`` to the operating system type to indicate that we don't have any additional
242	information.
243
244	Some examples:
245
246	If the Docker socket is available:
247	- Docker Desktop,5.15.90.1-microsoft-standard-WSL2,x86_64
248	- Linux Mint 21.1,5.19.0-32-generic,x86_64
249
250	If the Docker socket is not available, and we're on the host:
251	- Windows,10,AMD64
252	- Linux,5.19.0-32-generic,x86_64
253
254	If the Docker socket is not available, and we're in the container:
255	- Linux(Container),5.19.0-32-generic,x86_64
256
257	:return: A string representing the system's information
258	"""
259	try:	1✔
260	# try to get the system from the docker socket
261	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
262
263	system = DOCKER_CLIENT.get_system_info()	1✔
264
265	return ",".join(	1✔
266	[
267	system["OperatingSystem"],
268	system["KernelVersion"],
269	system["Architecture"],
270	]
271	)
NEW 272	except Exception:	×
NEW 273	if config.DEBUG_ANALYTICS:	×
NEW 274	LOG.exception(	×
275	"Unable to get system information from docker socket, falling back to platform.uname()"
276	)
277
NEW 278	uname = platform.uname()	×
279
NEW 280	if config.is_in_docker:	×
NEW 281	return ",".join(	×
282	[
283	f"{uname.system}(Container)",
284	uname.release,
285	uname.machine,
286	]
287	)
288
NEW 289	return ",".join(	×
290	[
291	uname.system,
292	uname.release,
293	uname.machine,
294	]
295	)
296
297
298	@hooks.prepare_host()	1✔
299	def prepare_host_machine_id():	1✔
300	# lazy-init machine ID into cache on the host, which can then be used in the container
301	get_machine_id()	1✔
302
303
304	@hooks.configure_localstack_container()	1✔
305	def _mount_machine_file(container: Container):	1✔
306	from localstack.utils.container_utils.container_client import BindMount	1✔
307
308	# mount tha machine file from the host's CLI cache directory into the appropriate location in the
309	# container
310	machine_file = os.path.join(config.dirs.cache, "machine.json")	1✔
311	if os.path.isfile(machine_file):	1✔
312	target = os.path.join(config.dirs.for_container().cache, "machine.json")	×
313	container.config.volumes.add(BindMount(machine_file, target, read_only=True))	×

localstack / localstack / 21697093787

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous