Read from envvar, then config file

Author: rw-access

detection_rules/eswrap.py

@@ -12,7 +12,7 @@
 from kibana import Kibana, RuleResource
 
 from .main import root
-from .misc import getenv
+from .misc import getdefault
 from .utils import normalize_timing_and_sort, unix_time_to_formatted, get_path
 from .rule_loader import get_rule, rta_mappings, load_rule_files, load_rules
 
@@ -179,10 +179,10 @@ def run(self, agent_hostname, indexes, verbose=True, **match):
 
 @es_group.command('collect-events')
 @click.argument('agent-hostname')
-@click.option('--elasticsearch-url', '-u', default=getenv("DR_ELASTICSEARCH_URL"))
-@click.option('--cloud-id', default=getenv("DR_ELASTICSEARCH_URL"))
-@click.option('--user', '-u', default=getenv("DR_USER"))
-@click.option('--password', '-p', default=getenv("DR_PASSWORD"), hide_input=True)
+@click.option('--elasticsearch-url', '-u', default=getdefault("elasticsearch_url"))
+@click.option('--cloud-id', default=getdefault("cloud_id"))
+@click.option('--user', '-u', default=getdefault("user"))
+@click.option('--password', '-p', default=getdefault("password"), hide_input=True)
 @click.option('--index', '-i', multiple=True, help='Index(es) to search against (default: all indexes)')
 @click.option('--agent-type', '-a', help='Restrict results to a source type (agent.type) ex: auditbeat')
 @click.option('--rta-name', '-r', help='Name of RTA in order to save events directly to unit tests data directory')
@@ -229,10 +229,10 @@ def normalize_file(events_file):
 
 @root.command("kibana-upload")
 @click.argument("toml-files", nargs=-1, required=True)
-@click.option('--kibana-url', '-u', default=getenv("DR_KIBANA_URL"))
-@click.option('--cloud-id', default=getenv("DR_CLOUD_ID"))
-@click.option('--user', '-u', default=getenv("DR_USER"))
-@click.option('--password', '-p', default=getenv("DR_PASSWORD"))
+@click.option('--kibana-url', '-u', default=getdefault("kibana_url"))
+@click.option('--cloud-id', default=getdefault("cloud_id"))
+@click.option('--user', '-u', default=getdefault("user"))
+@click.option('--password', '-p', default=getdefault("password"))
 def kibana_upload(toml_files, kibana_url, cloud_id, user, password):
     """Upload a list of rule .toml files to Kibana."""
     from uuid import uuid4

detection_rules/misc.py

@@ -12,7 +12,7 @@
 import click
 import requests
 
-from .utils import ROOT_DIR
+from .utils import ROOT_DIR, cached
 
 _CONFIG = {}
 
@@ -197,22 +197,23 @@ def download_worker(rule_info):
     return kibana_rules
 
 
+@cached
 def parse_config():
     """Parse a default config file."""
-    global _CONFIG
+    config_file = os.path.join(ROOT_DIR, '.detection-rules-cfg.json')
+    config = {}
 
-    if not _CONFIG:
-        config_file = os.path.join(ROOT_DIR, '.detection-rules-cfg.json')
+    if os.path.exists(config_file):
+        with open(config_file) as f:
+            config = json.load(f)
 
-        if os.path.exists(config_file):
-            with open(config_file) as f:
-                _CONFIG = json.load(f)
+        click.secho('Loaded config file: {}'.format(config_file), fg='yellow')
 
-            click.secho('Loaded config file: {}'.format(config_file), fg='yellow')
+    return config
 
-    return _CONFIG
 
-
-def getenv(name):
+def getdefault(name):
     """Callback function for `default` to get an environment variable."""
-    return lambda: os.environ.get(name)
+    envvar = f"DR_{name.upper()}"
+    config = parse_config()
+    return lambda: os.environ.get(envvar, config.get(name))