Changes positions.conf file path to /var/lib/intelmq/server/positions.json.

Author: gethvi

.reuse/dep5

@@ -86,4 +86,8 @@ License:   AGPL-3.0-or-later
 
 Files:     intelmq/etc/intelmq.yaml
 Copyright: 2023 Filip Pokorný
-License:   AGPL-3.0-or-later
+License:   AGPL-3.0-or-later
+
+Files:     intelmq/etc/positions.json
+Copyright: 2020 Birger Schacht, 2023 Filip Pokorný
+License:   CC0-1.0

debian/conffiles

@@ -1,3 +1,4 @@
 /etc/intelmq/harmonization.conf
 /etc/intelmq/runtime.yaml
 /etc/intelmq/intelmq.yaml
+/var/lib/intelmq/server/positions.json

debian/intelmq.install

@@ -7,3 +7,4 @@ intelmq/bots/experts/modify/examples/* usr/share/doc/intelmq/bots/experts/modify
 intelmq/etc/runtime.yaml etc/intelmq/
 intelmq/etc/harmonization.conf etc/intelmq/
 intelmq/etc/intelmq.yaml etc/intelmq/
+intelmq/etc/positions.json var/lib/intelmq/server/

intelmq/__init__.py

@@ -35,6 +35,7 @@
 HARMONIZATION_CONF_FILE = os.path.join(CONFIG_DIR, "harmonization.conf")
 RUNTIME_CONF_FILE = os.path.join(CONFIG_DIR, "runtime.yaml")
 INTELMQ_CONF_FILE = os.path.join(CONFIG_DIR, "intelmq.yaml")
+POSITIONS_FILE = os.path.join(VAR_SERVER_PATH, "positions.json")
 old_runtime_conf_file = pathlib.Path(RUNTIME_CONF_FILE).with_suffix('.conf')
 if not pathlib.Path(RUNTIME_CONF_FILE).exists() and old_runtime_conf_file.exists():
     old_runtime_conf_file.rename(RUNTIME_CONF_FILE)

intelmq/app/api/router.py

@@ -26,6 +26,7 @@
 import intelmq.app.api.runctl as runctl
 import intelmq.app.api.session as session
 
+from intelmq import RUNTIME_CONF_FILE, POSITIONS_FILE, HARMONIZATION_CONF_FILE
 from intelmq.app.dependencies import (app_config, cached_response, session_store,
                                       token_authorization)
 from .models import TokenResponse
@@ -193,10 +194,8 @@ def post_runtime(body: dict):
 
 @router.get("/positions", dependencies=[authorized], response_model=dict)
 def get_positions(runner: runctl.RunIntelMQCtl = Depends(runner)):
-    positions = pathlib.Path('/opt/intelmq/etc/manager/positions.conf')
     paths = runner.get_paths()
-    if 'CONFIG_DIR' in paths:
-        positions = pathlib.Path(paths['CONFIG_DIR']) / 'manager/positions.conf'
+    positions = pathlib.Path(paths.get("POSITIONS_FILE", POSITIONS_FILE))
     try:
         return json.loads(positions.read_text())
     except OSError as e:
@@ -207,10 +206,8 @@ def get_positions(runner: runctl.RunIntelMQCtl = Depends(runner)):
 @router.post("/positions", dependencies=[authorized], response_model=str,
              response_class=PlainTextResponse)
 def post_positions(body: dict, runner: runctl.RunIntelMQCtl = Depends(runner)):
-    positions = pathlib.Path('/opt/intelmq/etc/manager/positions.conf')
     paths = runner.get_paths()
-    if 'CONFIG_DIR' in paths:
-        positions = pathlib.Path(paths['CONFIG_DIR']) / 'manager/positions.conf'
+    positions = pathlib.Path(paths.get("POSITIONS_FILE", POSITIONS_FILE))
     try:
         positions.parent.mkdir(exist_ok=True)
         positions.write_text(json.dumps(body, indent=4))

intelmq/bin/intelmqctl.py

@@ -24,7 +24,7 @@
                      HARMONIZATION_CONF_FILE,
                      RUNTIME_CONF_FILE, VAR_RUN_PATH, STATE_FILE_PATH,
                      DEFAULT_LOGGING_PATH, __version_info__,
-                     CONFIG_DIR, ROOT_DIR)
+                     CONFIG_DIR, ROOT_DIR, POSITIONS_FILE, VAR_SERVER_PATH, INTELMQ_CONF_FILE)
 from intelmq.lib import utils
 from intelmq.lib.datatypes import ReturnType, MESSAGES, LogLevel
 from intelmq.lib.processmanager import *
@@ -1257,9 +1257,16 @@ def debug(self, sections=None):
             if self._returntype is ReturnType.TEXT:
                 print('Paths:')
             for path in ('HARMONIZATION_CONF_FILE',
-                         'RUNTIME_CONF_FILE', 'VAR_RUN_PATH', 'STATE_FILE_PATH',
-                         'DEFAULT_LOGGING_PATH', '__file__',
-                         'CONFIG_DIR', 'ROOT_DIR'):
+                         'RUNTIME_CONF_FILE',
+                         'INTELMQ_CONF_FILE',
+                         'POSITIONS_FILE',
+                         'VAR_RUN_PATH',
+                         'STATE_FILE_PATH',
+                         'DEFAULT_LOGGING_PATH',
+                         '__file__',
+                         'CONFIG_DIR',
+                         'ROOT_DIR',
+                         'VAR_SERVER_PATH'):
                 output['paths'][path] = variables[path]
                 if self._returntype is ReturnType.TEXT:
                     print(f'{path}: {variables[path]!r}')

intelmq/bin/intelmqsetup.py

@@ -94,19 +94,19 @@ def intelmqsetup_core(ownership=True, state_file=STATE_FILE_PATH):
 
     example_path = Path(pkg_resources.resource_filename('intelmq', 'etc'))
     example_confs = [
-        example_path / 'runtime.yaml',
-        example_path / 'harmonization.conf',
-        example_path / 'intelmq.yaml',
+        (example_path / 'runtime.yaml', Path(CONFIG_DIR) / 'runtime.yaml'),
+        (example_path / 'harmonization.conf', Path(CONFIG_DIR) / 'harmonization.conf'),
+        (example_path / 'intelmq.yaml', Path(CONFIG_DIR) / 'intelmq.yaml'),
+        (example_path / 'positions.json', Path(VAR_SERVER_PATH) / 'positions.json'),
     ]
-    for example_conf in example_confs:
+    for example_conf, destination_file in example_confs:
         fname = Path(example_conf).name
-        destination_file = Path(CONFIG_DIR) / fname
         if destination_file.exists():
             print(f'Not overwriting existing {fname!r} with example.')
             log_ownership_change = True
         else:
-            shutil.copy(example_conf, CONFIG_DIR)
-            print(f'Installing example {fname!r} to {CONFIG_DIR}.')
+            shutil.copy(example_conf, destination_file.parent)
+            print(f'Installing example {fname!r} to {destination_file.parent}.')
             log_ownership_change = False  # For installing the new files, we don't need to inform the admin that the permissions have been "fixed"
         if ownership:
             change_owner(destination_file, owner='intelmq', group='intelmq', log=log_ownership_change)

intelmq/etc/positions.json

@@ -0,0 +1,58 @@
+{
+    "feodo-tracker-browse-parser": {
+        "x": -304,
+        "y": 250
+    },
+    "feodo-tracker-browse-collector": {
+        "x": -508,
+        "y": 282
+    },
+    "cymru-whois-expert": {
+        "x": 510,
+        "y": -407
+    },
+    "deduplicator-expert": {
+        "x": -107,
+        "y": 162
+    },
+    "file-output": {
+        "x": 504,
+        "y": -614
+    },
+    "gethostbyname-1-expert": {
+        "x": 481,
+        "y": -198
+    },
+    "gethostbyname-2-expert": {
+        "x": 322,
+        "y": -325
+    },
+    "malc0de-parser": {
+        "x": -292,
+        "y": 48
+    },
+    "malc0de-windows-format-collector": {
+        "x": -477,
+        "y": -46
+    },
+    "spamhaus-drop-collector": {
+        "x": -88,
+        "y": 589
+    },
+    "spamhaus-drop-parser": {
+        "x": -114,
+        "y": 381
+    },
+    "taxonomy-expert": {
+        "x": 89,
+        "y": 29
+    },
+    "url2fqdn-expert": {
+        "x": 275,
+        "y": -116
+    },
+    "settings": {
+        "physics": false,
+        "live": true
+    }
+}

intelmq/tests/app/api/test_api.py

@@ -14,7 +14,6 @@
 from unittest.mock import patch, MagicMock
 from fastapi.testclient import TestClient
 
-
 with patch("intelmq.lib.utils.get_intelmq_settings", MagicMock(return_value={})):
 
     from intelmq.app import dependencies
@@ -97,7 +96,11 @@ def setUp(self) -> None:
         dependencies.startup(DummyConfig())
         self.conf_dir = TemporaryDirectory()
         app.dependency_overrides[runner] = get_dummy_reader(
-            paths={"CONFIG_DIR": self.conf_dir.name})
+            paths={
+                "CONFIG_DIR": self.conf_dir.name,
+                "POSITIONS_FILE": os.path.join(self.conf_dir.name, "positions.json")
+            }
+        )
 
         self.save_runtime()
         self.save_positions()
@@ -110,9 +113,12 @@ def save_runtime(self):
         with open(f"{self.conf_dir.name}/runtime.yaml", "w+") as f:
             json.dump({}, f)
 
+    def load_positions(self):
+        with open(f"{self.conf_dir.name}/positions.json") as f:
+            return json.load(f)
+
     def save_positions(self):
-        os.makedirs(f"{self.conf_dir.name}/manager", exist_ok=True)
-        with open(f"{self.conf_dir.name}/manager/positions.conf", "w+") as f:
+        with open(f"{self.conf_dir.name}/positions.json", "w+") as f:
             json.dump({}, f)
 
     def tearDown(self) -> None:
@@ -141,7 +147,6 @@ def test_post_runtime(self):
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.text, "success")
-
         self.assertEqual(utils.get_runtime(), data)
 
     def test_post_positions(self):
@@ -155,10 +160,7 @@ def test_post_positions(self):
 
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.text, "success")
-
-        with open(f"{self.conf_dir.name}/manager/positions.conf", "r") as f:
-            saved = json.load(f)
-        self.assertEqual(saved, data)
+        self.assertEqual(self.load_positions(), data)
 
 
 class TestAPILogin(TestCase):