# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one # or more contributor license agreements. Licensed under the Elastic License # 2.0; you may not use this file except in compliance with the Elastic License # 2.0. """Misc support.""" import os import unittest from collections.abc import Callable from functools import wraps from pathlib import Path from typing import IO, Any, NoReturn import click import requests from elasticsearch import AuthenticationException, Elasticsearch from kibana import Kibana # type: ignore[reportMissingTypeStubs] from .utils import add_params, cached, load_etc_dump LICENSE_HEADER = """ Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one or more contributor license agreements. Licensed under the Elastic License 2.0; you may not use this file except in compliance with the Elastic License 2.0. """.strip() LICENSE_LINES = LICENSE_HEADER.splitlines() PYTHON_LICENSE = "\n".join("# " + line for line in LICENSE_LINES) JS_LICENSE = """ /* {} */ """.strip().format("\n".join(" * " + line for line in LICENSE_LINES)) ROOT_DIR = Path(__file__).parent.parent class ClientError(click.ClickException): """Custom CLI error to format output or full debug stacktrace.""" def __init__(self, message: str, original_error: Exception | None = None) -> None: super().__init__(message) self.original_error = original_error self.original_error_type = type(original_error).__name__ if original_error else "" def show(self, file: IO[Any] | None = None, err: bool = True) -> None: """Print the error to the console.""" header = f"Error ({self.original_error_type})" msg = f"{click.style(header, fg='red', bold=True)}: {self.format_message()}" click.echo(msg, err=err, file=file) def raise_client_error( # noqa: PLR0913 message: str, exc: Exception | None = None, debug: bool | None = False, ctx: click.Context | None = None, file: IO[Any] | None = None, err: bool = False, ) -> NoReturn: config_debug = bool(ctx and ctx.ensure_object(dict) and ctx.obj.get("debug")) # type: ignore[reportUnknownArgumentType] debug = debug if debug is not None else config_debug if debug: click.echo(click.style("DEBUG: ", fg="yellow") + message, err=err, file=file) raise ClientError(message, original_error=exc) raise ClientError(message, original_error=exc) def nested_get(_dict: dict[str, Any] | None, dot_key: str | None, default: Any | None = None) -> Any: """Get a nested field from a nested dict with dot notation.""" if _dict is None or dot_key is None: return default if "." in dot_key: dot_key_parts = dot_key.split(".") this_key = dot_key_parts.pop(0) return nested_get(_dict.get(this_key, default), ".".join(dot_key_parts), default) return _dict.get(dot_key, default) def nested_set(_dict: dict[str, Any], dot_key: str, value: Any) -> None: """Set a nested field from a key in dot notation.""" keys = dot_key.split(".") for key in keys[:-1]: _dict = _dict.setdefault(key, {}) _dict[keys[-1]] = value def nest_from_dot(dots: str, value: Any) -> Any: """Nest a dotted field and set the innermost value.""" fields = dots.split(".") if not fields: return {} nested = {fields.pop(): value} for field_ in reversed(fields): nested = {field_: nested} return nested def get_kibana_rules_map(repo: str = "elastic/kibana", branch: str = "master") -> dict[str, Any]: """Get list of available rules from the Kibana repo and return a list of URLs.""" timeout = 30 # secs # ensure branch exists r = requests.get(f"https://api.github.com/repos/{repo}/branches/{branch}", timeout=timeout) r.raise_for_status() url = ( "https://api.github.com/repos/{repo}/contents/x-pack/{legacy}plugins/{app}/server/lib/" "detection_engine/rules/prepackaged_rules?ref={branch}" ) r = requests.get(url.format(legacy="", app="security_solution", branch=branch, repo=repo), timeout=timeout) r.raise_for_status() gh_rules = r.json() # pre-7.9 app was siem if isinstance(gh_rules, dict) and gh_rules.get("message", "") == "Not Found": # type: ignore[reportUnknownMemberType] gh_rules = requests.get(url.format(legacy="", app="siem", branch=branch, repo=repo), timeout=timeout).json() # pre-7.8 the siem was under the legacy directory if isinstance(gh_rules, dict) and gh_rules.get("message", "") == "Not Found": # type: ignore[reportUnknownMemberType] gh_rules = requests.get( url.format(legacy="legacy/", app="siem", branch=branch, repo=repo), timeout=timeout ).json() if isinstance(gh_rules, dict) and gh_rules.get("message", "") == "Not Found": # type: ignore[reportUnknownMemberType] raise ValueError(f"rules directory does not exist for {repo} branch: {branch}") if not isinstance(gh_rules, list): raise TypeError("Expected to receive a list") results: dict[str, Any] = {} for r in gh_rules: # type: ignore[reportUnknownMemberType] if "name" not in r: raise ValueError("Name value is expected") name = r["name"] # type: ignore[reportUnknownMemberType] if not isinstance(name, str): raise TypeError("String value is expected for name") if name.endswith(".json"): key = Path(name).name val = r["download_url"] # type: ignore[reportUnknownMemberType] results[key] = val return results def get_kibana_rules( repo: str = "elastic/kibana", branch: str = "master", verbose: bool = True, threads: int = 50, rule_paths: list[str] | None = None, ) -> dict[str, Any]: """Retrieve prepackaged rules from kibana repo.""" from multiprocessing.pool import ThreadPool kibana_rules: dict[str, Any] = {} if verbose: thread_use = f" using {threads} threads" if threads > 1 else "" click.echo(f"Downloading rules from {repo} {branch} branch in kibana repo{thread_use} ...") rule_paths = [os.path.splitext(os.path.basename(p))[0] for p in (rule_paths or [])] # noqa: PTH119, PTH122 rules_mapping = ( [(n, u) for n, u in get_kibana_rules_map(repo=repo, branch=branch).items() if n in rule_paths] if rule_paths else get_kibana_rules_map(repo=repo, branch=branch).items() ) def download_worker(rule_info: tuple[str, str]) -> None: n, u = rule_info kibana_rules[n] = requests.get(u, timeout=30).json() pool = ThreadPool(processes=threads) _ = pool.map(download_worker, rules_mapping) pool.close() pool.join() return kibana_rules @cached def load_current_package_version() -> str: """Load the current package version from config file.""" data = load_etc_dump(["packages.yaml"]) return data["package"]["name"] def get_default_config() -> Path | None: return next(ROOT_DIR.glob(".detection-rules-cfg.*"), None) @cached def parse_user_config() -> dict[str, Any]: """Parse a default config file.""" import eql # type: ignore[reportMissingTypeStubs] config_file = get_default_config() config = {} if config_file and config_file.exists(): config = eql.utils.load_dump(str(config_file)) # type: ignore[reportUnknownMemberType] click.secho(f"Loaded config file: {config_file}", fg="yellow") return config def discover_tests(start_dir: str = "tests", pattern: str = "test*.py", top_level_dir: str | None = None) -> list[str]: """Discover all unit tests in a directory.""" tests: list[str] = [] def list_tests(s: unittest.TestSuite) -> None: for test in s: if isinstance(test, unittest.TestSuite): list_tests(test) else: tests.append(test.id()) loader = unittest.defaultTestLoader suite = loader.discover(start_dir, pattern=pattern, top_level_dir=top_level_dir or str(ROOT_DIR)) list_tests(suite) return tests def getdefault(name: str) -> Callable[[], Any]: """Callback function for `default` to get an environment variable.""" envvar = f"DR_{name.upper()}" config = parse_user_config() return lambda: os.environ.get(envvar, config.get(name)) def get_elasticsearch_client( # noqa: PLR0913 cloud_id: str | None = None, elasticsearch_url: str | None = None, es_user: str | None = None, es_password: str | None = None, ctx: click.Context | None = None, api_key: str | None = None, **kwargs: Any, ) -> Elasticsearch: """Get an authenticated elasticsearch client.""" # Handle empty strings as None cloud_id = cloud_id or None elasticsearch_url = elasticsearch_url or None if not (cloud_id or elasticsearch_url): raise_client_error("Missing required --cloud-id or --elasticsearch-url") # don't prompt for these until there's a cloud id or elasticsearch URL basic_auth: tuple[str, str] | None = None if not api_key: es_user = es_user or click.prompt("es_user") es_password = es_password or click.prompt("es_password", hide_input=True) if not es_user or not es_password: raise ValueError("Both username and password must be provided") basic_auth = (es_user, es_password) hosts = [elasticsearch_url] if elasticsearch_url else None timeout = kwargs.pop("timeout", 60) kwargs["verify_certs"] = not kwargs.pop("ignore_ssl_errors", False) try: client = Elasticsearch( hosts=hosts, cloud_id=cloud_id, http_auth=basic_auth, timeout=timeout, api_key=api_key, **kwargs ) # force login to test auth _ = client.info() except AuthenticationException as e: error_msg = f"Failed authentication for {elasticsearch_url or cloud_id}" raise_client_error(error_msg, e, ctx=ctx, err=True) else: return client def get_default_elasticsearch_client() -> Elasticsearch: """Get an default authenticated elasticsearch client.""" return get_elasticsearch_client( api_key=getdefault("api_key")(), cloud_id=getdefault("cloud_id")(), elasticsearch_url=getdefault("elasticsearch_url")(), ignore_ssl_errors=getdefault("ignore_ssl_errors")(), ) def get_kibana_client( *, api_key: str, cloud_id: str | None = None, kibana_url: str | None = None, space: str | None = None, ignore_ssl_errors: bool = False, **kwargs: Any, ) -> Kibana: """Get an authenticated Kibana client.""" if not (cloud_id or kibana_url): raise_client_error("Missing required --cloud-id or --kibana-url") verify = not ignore_ssl_errors return Kibana(cloud_id=cloud_id, kibana_url=kibana_url, space=space, verify=verify, api_key=api_key, **kwargs) def get_default_kibana_client() -> Kibana: """Get a default authenticated Kibana client.""" return get_kibana_client( api_key=getdefault("api_key")(), cloud_id=getdefault("cloud_id")(), kibana_url=getdefault("kibana_url")(), space=getdefault("space")(), ignore_ssl_errors=getdefault("ignore_ssl_errors")(), ) client_options = { "kibana": { "kibana_url": click.Option(["--kibana-url"], default=getdefault("kibana_url")), "cloud_id": click.Option(["--cloud-id"], default=getdefault("cloud_id"), help="ID of the cloud instance."), "api_key": click.Option(["--api-key"], default=getdefault("api_key")), "space": click.Option(["--space"], default=None, help="Kibana space"), "ignore_ssl_errors": click.Option(["--ignore-ssl-errors"], default=getdefault("ignore_ssl_errors")), }, "elasticsearch": { "cloud_id": click.Option(["--cloud-id"], default=getdefault("cloud_id")), "api_key": click.Option(["--api-key"], default=getdefault("api_key")), "elasticsearch_url": click.Option(["--elasticsearch-url"], default=getdefault("elasticsearch_url")), "es_user": click.Option(["--es-user", "-eu"], default=getdefault("es_user")), "es_password": click.Option(["--es-password", "-ep"], default=getdefault("es_password")), "timeout": click.Option(["--timeout", "-et"], default=60, help="Timeout for elasticsearch client"), "ignore_ssl_errors": click.Option(["--ignore-ssl-errors"], default=getdefault("ignore_ssl_errors")), }, } kibana_options = list(client_options["kibana"].values()) elasticsearch_options = list(client_options["elasticsearch"].values()) def add_client(client_types: list[str], add_to_ctx: bool = True, add_func_arg: bool = True) -> Callable[..., Any]: """Wrapper to add authed client.""" def _wrapper(func: Callable[..., Any]) -> Callable[..., Any]: client_ops_dict: dict[str, click.Option] = {} client_ops_keys: dict[str, list[str]] = {} for c_type in client_types: ops = client_options[c_type] client_ops_dict.update(ops) client_ops_keys[c_type] = list(ops) if not client_ops_dict: client_types_str = ", ".join(client_types) raise ValueError(f"Unknown client: {client_types_str} in {func.__name__}") client_ops = list(client_ops_dict.values()) @wraps(func) @add_params(*client_ops) def _wrapped(*args: Any, **kwargs: Any) -> Any: # noqa: PLR0912 ctx: click.Context | None = next((a for a in args if isinstance(a, click.Context)), None) es_client_args = {k: kwargs.pop(k, None) for k in client_ops_keys.get("elasticsearch", [])} # shared args like cloud_id kibana_client_args = {k: kwargs.pop(k, es_client_args.get(k)) for k in client_ops_keys.get("kibana", [])} if "elasticsearch" in client_types: # for nested ctx invocation, no need to re-auth if an existing client is already passed elasticsearch_client: Elasticsearch | None = kwargs.get("elasticsearch_client") try: if elasticsearch_client and elasticsearch_client.info(): pass else: elasticsearch_client = get_elasticsearch_client(**es_client_args) except AuthenticationException: elasticsearch_client = get_elasticsearch_client(**es_client_args) if add_func_arg: kwargs["elasticsearch_client"] = elasticsearch_client if ctx and add_to_ctx: ctx.obj["es"] = elasticsearch_client if "kibana" in client_types: # for nested ctx invocation, no need to re-auth if an existing client is already passed kibana_client: Kibana | None = kwargs.get("kibana_client") if kibana_client: try: with kibana_client: if kibana_client.version: pass # kibana_client is valid and can be used directly except (requests.HTTPError, AttributeError): kibana_client = get_kibana_client(**kibana_client_args) else: # Instantiate a new Kibana client if none was provided or if the provided one is not usable kibana_client = get_kibana_client(**kibana_client_args) if add_func_arg: kwargs["kibana_client"] = kibana_client if ctx and add_to_ctx: ctx.obj["kibana"] = kibana_client return func(*args, **kwargs) return _wrapped return _wrapper