intelmq.lib.mixins package

Submodules

intelmq.lib.mixins.cache module

CacheMixin for IntelMQ

SPDX-FileCopyrightText: 2021 Sebastian Waldbauer SPDX-License-Identifier: AGPL-3.0-or-later

CacheMixin is used for caching/storing data in redis.

class intelmq.lib.mixins.cache.CacheMixin(**kwargs)

Bases: object

cache_exists(key: str)

cache_flush()

Flushes the currently opened database by calling FLUSHDB.

cache_get(key: str)

cache_get_redis_instance()

cache_set(key: str, value: Any, ttl: int | None = None)

redis_cache_db: int = 9

redis_cache_host: str = '127.0.0.1'

redis_cache_password: str | None = None

redis_cache_port: int = 6379

redis_cache_ttl: int = 15

intelmq.lib.mixins.http module

HttpMixin for IntelMQ

SPDX-FileCopyrightText: 2021 Birger Schacht SPDX-License-Identifier: AGPL-3.0-or-later

Based on create_request_session in intelmq.lib.utils and set_request_parameters in intelmq.lib.bot.Bot

class intelmq.lib.mixins.http.HttpMixin(**kwargs)

Bases: object

Setup a request session

http_get(url: str, **kwargs) → Response

http_header: dict = {}

http_password = None

http_proxy = None

http_session() → Session

http_timeout_max_tries: int = 3

http_timeout_sec: int = 30

http_user_agent: str = 'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36'

http_username = None

http_verify_cert: bool = True

https_proxy = None

setup()

ssl_client_cert = None

class intelmq.lib.mixins.http.TimeoutHTTPAdapter(*args, timeout=None, **kwargs)

Bases: HTTPAdapter

A requests-HTTP Adapter which can set the timeout generally.

send(*args, **kwargs)

Sends PreparedRequest object. Returns Response object.

Parameters:

• request – The PreparedRequest being sent.

• stream – (optional) Whether to stream the request content.

• timeout (float or tuple or urllib3 Timeout object) – (optional) How long to wait for the server to send data before giving up, as a float, or a (connect timeout, read timeout) tuple.

• verify – (optional) Either a boolean, in which case it controls whether we verify the server’s TLS certificate, or a string, in which case it must be a path to a CA bundle to use

• cert – (optional) Any user-provided SSL certificate to be trusted.

• proxies – (optional) The proxies dictionary to apply to the request.

Return type:

requests.Response

intelmq.lib.mixins.sql module

SQLMixin for IntelMQ

SPDX-FileCopyrightText: 2021 Birger Schacht, 2022 Intevation GmbH SPDX-License-Identifier: AGPL-3.0-or-later

Based on the former SQLBot base class

class intelmq.lib.mixins.sql.SQLMixin(*args, **kwargs)

Bases: object

Inherit this bot so that it handles DB connection for you. You do not have to bother: * connecting database in the self.init() method, just call super().init(), self.cur will be set * catching exceptions, just call self.execute() instead of self.cur.execute() * self.format_char will be set to ‘%s’ in PostgreSQL and to ‘?’ in SQLite

MSSQL = 'mssql'

POSTGRESQL = 'postgresql'

SQLITE = 'sqlite'

engine = None

execute(query: str, values: tuple, rollback=False)

fail_on_errors = False

message_jsondict_as_string = True

reconnect_delay = 0

intelmq.lib.mixins.stomp module

StompMixin for IntelMQ

SPDX-FileCopyrightText: 2017 Sebastian Wagner, 2023 NASK SPDX-License-Identifier: AGPL-3.0-or-later

class intelmq.lib.mixins.stomp.StompMixin

Bases: object

A mixin that provides certain common methods for STOMP bots.

auth_by_ssl_client_certificate: bool

heartbeat: int

password: str

port: int

prepare_stomp_connection() → Tuple[stomp.Connection, dict]

Get a (<STOMP connection>, <STOMP connect arguments>) pair.

• <STOMP connection> is a new instance of stomp.Connection,

  with the SSL stuff already configured, but without any invocations of connect() made yet;

• <STOMP connect arguments> is a dict of arguments – ready to be passed to the connect() method of the aforementioned <STOMP connection> object.

server: str

ssl_ca_certificate: str

ssl_client_certificate: str

ssl_client_certificate_key: str

classmethod stomp_bot_parameters_check(parameters: dict) → List[List[str]]

Intended to be used in bots’ check() static/class method.

stomp_bot_runtime_initial_check() → None

Intended to be used in bots’ init() instance method.

username: str

Module contents

class intelmq.lib.mixins.CacheMixin(**kwargs)

Bases: object

cache_exists(key: str)

cache_flush()

Flushes the currently opened database by calling FLUSHDB.

cache_get(key: str)

cache_get_redis_instance()

cache_set(key: str, value: Any, ttl: int | None = None)

redis_cache_db: int = 9

redis_cache_host: str = '127.0.0.1'

redis_cache_password: str | None = None

redis_cache_port: int = 6379

redis_cache_ttl: int = 15

class intelmq.lib.mixins.HttpMixin(**kwargs)

Bases: object

Setup a request session

http_get(url: str, **kwargs) → Response

http_header: dict = {}

http_password = None

http_proxy = None

http_session() → Session

http_timeout_max_tries: int = 3

http_timeout_sec: int = 30

http_user_agent: str = 'Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36'

http_username = None

http_verify_cert: bool = True

https_proxy = None

setup()

ssl_client_cert = None

class intelmq.lib.mixins.SQLMixin(*args, **kwargs)

Bases: object

Inherit this bot so that it handles DB connection for you. You do not have to bother: * connecting database in the self.init() method, just call super().init(), self.cur will be set * catching exceptions, just call self.execute() instead of self.cur.execute() * self.format_char will be set to ‘%s’ in PostgreSQL and to ‘?’ in SQLite

MSSQL = 'mssql'

POSTGRESQL = 'postgresql'

SQLITE = 'sqlite'

engine = None

execute(query: str, values: tuple, rollback=False)

fail_on_errors = False

message_jsondict_as_string = True

reconnect_delay = 0

class intelmq.lib.mixins.StompMixin

Bases: object

A mixin that provides certain common methods for STOMP bots.

auth_by_ssl_client_certificate: bool

heartbeat: int

password: str

port: int

prepare_stomp_connection() → Tuple[stomp.Connection, dict]

Get a (<STOMP connection>, <STOMP connect arguments>) pair.

• <STOMP connection> is a new instance of stomp.Connection,

  with the SSL stuff already configured, but without any invocations of connect() made yet;

• <STOMP connect arguments> is a dict of arguments – ready to be passed to the connect() method of the aforementioned <STOMP connection> object.

server: str

ssl_ca_certificate: str

ssl_client_certificate: str

ssl_client_certificate_key: str

classmethod stomp_bot_parameters_check(parameters: dict) → List[List[str]]

Intended to be used in bots’ check() static/class method.

stomp_bot_runtime_initial_check() → None

Intended to be used in bots’ init() instance method.

username: str