Merges API and Manager to IntelMQ project.

intelmq/__main__.py

@@ -0,0 +1,72 @@
+import argparse
+import getpass
+import sys
+
+import uvicorn
+
+from intelmq.api.config import Config
+from intelmq.api.session import SessionStore
+from intelmq.lib import utils
+
+
+def server_start(host: str = None, port: int = None, debug: bool = False, *args, **kwargs):
+    server_settings = utils.get_server_settings()
+    host = host if host is not None else server_settings.get("host", "0.0.0.0")
+    port = int(port) if port is not None else int(server_settings.get("port", 8080))
+
+    return uvicorn.run(
+        "intelmq.server:app",
+        host=host,
+        reload=debug,
+        port=port,
+        workers=1,
+    )
+
+
+def server_adduser(username: str, password: str = None, *args, **kwargs):
+    api_config: Config = Config()
+
+    if api_config.session_store is None:
+        print("Could not add user- no session store configured in configuration!", file=sys.stderr)
+        exit(1)
+
+    session_store = SessionStore(str(api_config.session_store), api_config.session_duration)
+    password = getpass.getpass() if password is None else password
+    session_store.add_user(username, password)
+    print(f"Added user {username} to intelmq session file.")
+
+
+def main():
+    parser = argparse.ArgumentParser(prog="intelmq", usage="intelmq [OPTIONS] COMMAND")
+    parser.set_defaults(func=(lambda *_, **__: parser.print_help())) # wrapper to accept args and kwargs
+    parser._optionals.title = "Options"
+    parser.add_argument("--version", action="store_true", help="print version and exit", default=None)
+    commands = parser.add_subparsers(metavar="", title="Commands")
+
+    # intelmq server
+    srv_parser = commands.add_parser("server", help="server subcommands", usage="intelmq server [COMMAND]")
+    srv_parser.set_defaults(func=(lambda *_, **__: srv_parser.print_help())) # wrapper to accept args and kwargs
+    srv_parser._optionals.title = "Options"
+    srv_subcommands = srv_parser.add_subparsers(metavar="", title="Commands")
+
+    # intelmq server start
+    srv_start = srv_subcommands.add_parser("start", help="start the server", usage="intelmq server start [OPTIONS]")
+    srv_start.set_defaults(func=server_start)
+    srv_start._optionals.title = "Options"
+    srv_start.add_argument("--debug", action="store_true", dest="debug", default=None)
+    srv_start.add_argument("--host", type=str, dest="host")
+    srv_start.add_argument("--port", type=int, dest="port")
+
+    # intelmq server adduser
+    srv_adduser = srv_subcommands.add_parser("adduser", help="adds new user", usage="intelmq server adduser [OPTIONS]")
+    srv_adduser.set_defaults(func=server_adduser)
+    srv_adduser._optionals.title = "Options"
+    srv_adduser.add_argument('--username', required=True, help='The username of the account.', type=str)
+    srv_adduser.add_argument('--password', required=False, help='The password of the account.', type=str)
+
+    args = parser.parse_args()
+    return args.func(**vars(args))
+
+
+if __name__ == "__main__":
+    main()

intelmq/api/config.py

@@ -0,0 +1,61 @@
+"""Configuration for IntelMQ Manager
+
+SPDX-FileCopyrightText: 2020 Intevation GmbH <https://intevation.de>
+SPDX-License-Identifier: AGPL-3.0-or-later
+
+Funding: of initial version by SUNET
+Author(s):
+  * Bernhard Herzog <[email protected]>
+"""
+
+from typing import List, Optional
+from pathlib import Path
+from intelmq.lib import utils
+
+
+class Config:
+
+    """Configuration settings for IntelMQ Manager"""
+
+    intelmq_ctl_cmd: List[str] = ["sudo", "-u", "intelmq", "/usr/local/bin/intelmqctl"]
+
+    allowed_path: Path = Path("/opt/intelmq/var/lib/bots/")
+
+    session_store: Optional[Path] = None
+
+    session_duration: int = 24 * 3600
+
+    allow_origins: List[str] = ['*']
+
+    enable_webgui: bool = True
+
+    host: str = "0.0.0.0"
+
+    port: int = 8080
+
+    def __init__(self):
+        server_settings = utils.get_server_settings()
+
+        if "intelmq_ctl_cmd" in server_settings:
+            self.intelmq_ctl_cmd = server_settings["intelmq_ctl_cmd"]
+
+        if "allowed_path" in server_settings:
+            self.allowed_path = Path(server_settings["allowed_path"])
+
+        if "session_store" in server_settings:
+            self.session_store = Path(server_settings["session_store"])
+
+        if "session_duration" in server_settings:
+            self.session_duration = int(server_settings["session_duration"])
+
+        if "allow_origins" in server_settings:
+            self.allow_origins = server_settings['allow_origins']
+
+        if "enable_webgui" in server_settings:
+            self.enable_webgui = server_settings["enable_webgui"]
+
+        if "host" in server_settings:
+            self.host = server_settings["host"]
+
+        if "port" in server_settings:
+            self.host = server_settings["port"]

intelmq/api/dependencies.py

@@ -0,0 +1,62 @@
+"""Dependencies of the API endpoints, in the FastAPI style
+
+SPDX-FileCopyrightText: 2022 CERT.at GmbH <https://cert.at>
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+
+import typing
+from typing import Generic, Optional, TypeVar
+
+from fastapi import Depends, Header, HTTPException, Response, status
+
+import intelmq.api.config
+import intelmq.api.session as session
+
+T = TypeVar("T")
+
+
+class OneTimeDependency(Generic[T]):
+    """Allows one-time explicit initialization of the dependency,
+        and then returning it on every usage.
+
+        It emulates the previous behavior that used global variables"""
+
+    def __init__(self) -> None:
+        self._value: Optional[T] = None
+
+    def initialize(self, value: T) -> None:
+        self._value = value
+
+    def __call__(self) -> Optional[T]:
+        return self._value
+
+
+api_config = OneTimeDependency[intelmq.api.config.Config]()
+session_store = OneTimeDependency[session.SessionStore]()
+
+
+def cached_response(max_age: int):
+    """Adds the cache headers to the response"""
+    def _cached_response(response: Response):
+        response.headers["cache-control"] = f"max-age={max_age}"
+    return _cached_response
+
+
+def token_authorization(authorization: typing.Union[str, None] = Header(default=None),
+                        session: session.SessionStore = Depends(session_store)):
+    if session is not None:
+        if not authorization or not session.verify_token(authorization):
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail={
+                                "Authentication Required":
+                                    "Please provide valid Token verification credentials"
+                                })
+
+
+def startup(config: intelmq.api.config.Config):
+    """A starting point to one-time initialization of necessary dependencies. This needs to
+        be called by the application on the startup."""
+    api_config.initialize(config)
+    session_file = config.session_store
+    if session_file is not None:
+        session_store.initialize(session.SessionStore(str(session_file),
+                                                      config.session_duration))

intelmq/api/exceptions.py

@@ -0,0 +1,25 @@
+"""Exception handlers for API
+
+SPDX-FileCopyrightText: 2022 CERT.at GmbH <https://cert.at>
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+
+from fastapi import FastAPI, Request, status
+from fastapi.responses import JSONResponse
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+import intelmq.api.runctl as runctl
+
+
+def ctl_error_handler(request: Request, exc: runctl.IntelMQCtlError):
+    return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=exc.error_dict)
+
+
+def handle_generic_error(request: Request, exc: StarletteHTTPException):
+    return JSONResponse(status_code=exc.status_code, content={"error": exc.detail})
+
+
+def register(app: FastAPI):
+    """A hook to register handlers in the app. Need to be called before startup"""
+    app.add_exception_handler(runctl.IntelMQCtlError, ctl_error_handler)
+    app.add_exception_handler(StarletteHTTPException, handle_generic_error)

intelmq/api/files.py

@@ -0,0 +1,79 @@
+"""Direct access to IntelMQ files and directories
+
+SPDX-FileCopyrightText: 2020 Intevation GmbH <https://intevation.de>
+SPDX-License-Identifier: AGPL-3.0-or-later
+
+Funding: of initial version by SUNET
+Author(s):
+  * Bernhard Herzog <[email protected]>
+
+This module implements the part of the IntelMQ-Manager backend that
+allows direct read and write access to some of the files used by
+IntelMQ.
+"""
+
+from pathlib import PurePath, Path
+from typing import Optional, Tuple, Union, Dict, Any, Iterable, BinaryIO
+
+from intelmq.api.config import Config
+
+
+def path_starts_with(path: PurePath, prefix: PurePath) -> bool:
+    """Return whether the path starts with prefix.
+
+    Both arguments must be absolute paths. If not, this function raises
+    a ValueError.
+
+    This function compares the path components, so it's not a simple
+    string prefix test.
+    """
+    if not path.is_absolute():
+        raise ValueError("{!r} is not absolute".format(path))
+    if not prefix.is_absolute():
+        raise ValueError("{!r} is not absolute".format(prefix))
+    return path.parts[:len(prefix.parts)] == prefix.parts
+
+
+class FileAccess:
+
+    def __init__(self, config: Config):
+        self.allowed_path = config.allowed_path
+
+    def file_name_allowed(self, filename: str) -> Optional[Tuple[bool, Path]]:
+        """Determine wether the API should allow access to a file."""
+        resolved = Path(filename).resolve()
+        if not path_starts_with(resolved, self.allowed_path):
+            return None
+
+        return (False, resolved)
+
+    def load_file_or_directory(self, unvalidated_filename: str, fetch: bool) \
+            -> Union[Tuple[str, Union[BinaryIO, Dict[str, Any]]], None]:
+        allowed = self.file_name_allowed(unvalidated_filename)
+        if allowed is None:
+            return None
+
+        content_type = "application/json"
+        predefined, normalized = allowed
+
+        if predefined or fetch:
+            if fetch:
+                content_type = "text/html"
+            return (content_type, open(normalized, "rb"))
+
+        result = {"files": {}}  # type: Dict[str, Any]
+        if normalized.is_dir():
+            result["directory"] = str(normalized)
+            files = normalized.iterdir()  # type: Iterable[Path]
+        else:
+            files = [normalized]
+
+        for path in files:
+            stat = path.stat()
+            if stat.st_size < 2000:
+                # FIXME: don't hardwire this size
+                obj = {"contents": path.read_text()}  # type: Dict[str, Any]
+            else:
+                obj = {"size": stat.st_size, "path": str(path.resolve())}
+            result["files"][path.name] = obj
+        return (content_type, result)

intelmq/api/models.py

@@ -0,0 +1,12 @@
+"""Models used in API
+
+SPDX-FileCopyrightText: 2023 CERT.at GmbH <https://cert.at/>
+SPDX-License-Identifier: AGPL-3.0-or-later
+"""
+
+from pydantic import BaseModel
+
+
+class TokenResponse(BaseModel):
+    login_token: str
+    username: str