788218a1-7c88-4b9b-95b7-0b9b914593b4

Committed 10 Feb 2025 11:28PM UTC coverage: 86.929% (+0.007%) from 86.922%

Build # 788218a1-7c88-4b9b-95b7-0b9b914593b4

Build Type

push

circleci

Committed by

web-flow

Commit Message

Step Functions: Add Supported Integrations For Glue (#12248)

Coverage Stats

2 of 3 new or added lines in 1 file covered. (66.67%)

124 existing lines in 10 files now uncovered.

61509 of 70758 relevant lines covered (86.93%)

0.87 hits per line

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

69.92

/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.exception("Could not determine license activation status")
        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 VolumeBind

    # 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(VolumeBind(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✔
UNCOV 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✔
UNCOV 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
UNCOV 99	try:	×
UNCOV 100	doc = FileMappedDocument(cache_path)	×
UNCOV 101	except Exception:	×
UNCOV 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
UNCOV 146	try:	×
UNCOV 147	from localstack.pro.core.bootstrap import licensingv2	×
148
UNCOV 149	return licensingv2.get_licensed_environment().activated	×
UNCOV 150	except Exception:	×
UNCOV 151	LOG.exception("Could not determine license activation status")	×
UNCOV 152	return False	×
153
154
155	def _generate_session_id() -> str:	1✔
156	return long_uid()	1✔
157
158
159	def _anonymize_physical_id(physical_id: str) -> str:	1✔
160	"""
161	Returns 12 digits of the salted hash of the given physical ID.
162
163	:param physical_id: the physical id
164	:return: an anonymized 12 digit value representing the physical ID.
165	"""
166	hashed = md5(_PHYSICAL_ID_SALT + physical_id)	1✔
167	return hashed[:12]	1✔
168
169
170	def _generate_machine_id() -> str:	1✔
171	try:	1✔
172	# try to get a robust ID from the docker socket (which will be the same from the host and the
173	# container)
174	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
175
176	docker_id = DOCKER_CLIENT.get_system_id()	1✔
177	# some systems like podman don't return a stable ID, so we double-check that here
178	if docker_id == DOCKER_CLIENT.get_system_id():	1✔
179	return f"dkr_{_anonymize_physical_id(docker_id)}"	1✔
UNCOV 180	except Exception:	×
UNCOV 181	pass	×
182
UNCOV 183	if config.is_in_docker:	×
UNCOV 184	return f"gen_{long_uid()[:12]}"	×
185
186	# this can potentially be useful when generated on the host using the CLI and then mounted into the
187	# container via machine.json
UNCOV 188	try:	×
189	if os.path.exists("/etc/machine-id"):	×
UNCOV 190	with open("/etc/machine-id") as fd:	×
UNCOV 191	machine_id = str(fd.read()).strip()	×
UNCOV 192	if machine_id:	×
193	return f"sys_{_anonymize_physical_id(machine_id)}"	×
UNCOV 194	except Exception:	×
UNCOV 195	pass	×
196
197	# always fall back to a generated id
UNCOV 198	return f"gen_{long_uid()[:12]}"	×
199
200
201	def get_api_key_or_auth_token() -> Optional[str]:	1✔
202	# TODO: this is duplicated code from ext, but should probably migrate that to localstack
203	auth_token = os.environ.get("LOCALSTACK_AUTH_TOKEN", "").strip("'\" ")	1✔
204	if auth_token:	1✔
UNCOV 205	return auth_token	×
206
207	api_key = os.environ.get("LOCALSTACK_API_KEY", "").strip("'\" ")	1✔
208	if api_key:	1✔
UNCOV 209	return api_key	×
210
211	return None	1✔
212
213
214	@singleton_factory	1✔
215	def get_system() -> str:	1✔
216	try:	1✔
217	# try to get the system from the docker socket
218	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
219
220	system = DOCKER_CLIENT.get_system_info()	1✔
221	if system.get("OSType"):	1✔
222	return system.get("OSType").lower()	1✔
UNCOV 223	except Exception:	×
224	pass	×
225
UNCOV 226	if config.is_in_docker:	×
UNCOV 227	return "docker"	×
228
229	return platform.system().lower()	×
230
231
232	@hooks.prepare_host()	1✔
233	def prepare_host_machine_id():	1✔
234	# lazy-init machine ID into cache on the host, which can then be used in the container
235	get_machine_id()	1✔
236
237
238	@hooks.configure_localstack_container()	1✔
239	def _mount_machine_file(container: Container):	1✔
UNCOV 240	from localstack.utils.container_utils.container_client import VolumeBind	×
241
242	# mount tha machine file from the host's CLI cache directory into the appropriate location in the
243	# container
UNCOV 244	machine_file = os.path.join(config.dirs.cache, "machine.json")	×
UNCOV 245	if os.path.isfile(machine_file):	×
UNCOV 246	target = os.path.join(config.dirs.for_container().cache, "machine.json")	×
UNCOV 247	container.config.volumes.add(VolumeBind(machine_file, target, read_only=True))	×

localstack / localstack / 788218a1-7c88-4b9b-95b7-0b9b914593b4

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