16981563750

Committed 14 Aug 2025 10:49PM UTC coverage: 86.896% (+0.04%) from 86.852%

Build # 16981563750

Build Type

push

github

Committed by

web-flow

Commit Message

add support for Fn::Tranform in CFnV2 (#12966)

Co-authored-by: Simon Walker <simon.walker@localstack.cloud>

Coverage Stats

181 of 195 new or added lines in 6 files covered. (92.82%)

348 existing lines in 22 files now uncovered.

66915 of 77006 relevant lines covered (86.9%)

0.87 hits per line

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

72.18

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

import dataclasses
import logging
import os
import platform
from typing import Optional

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 "ClientMetadata(%s)" % 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(),
        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() -> Optional[str]:
    # 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:
    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()


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

localstack / localstack / 16981563750

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