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: typing.Any, ttl: typing.Union[int, NoneType] = None)
redis_cache_db = 9
redis_cache_host = '127.0.0.1'
redis_cache_password = None
redis_cache_port = 6379
redis_cache_ttl = 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) → requests.models.Response
http_header = {}
http_password = None
http_proxy = None
http_session() → requests.sessions.Session
http_timeout_max_tries = 3
http_timeout_sec = 30
http_user_agent = '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 = True
https_proxy = None
setup()
ssl_client_cert = None
class intelmq.lib.mixins.http.TimeoutHTTPAdapter(*args, timeout=None, **kwargs)

Bases: requests.adapters.HTTPAdapter

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

send(*args, **kwargs)

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)
message_jsondict_as_string = True
reconnect_delay = 0

Module contents

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

Bases: object

Setup a request session

http_get(url: str, **kwargs) → requests.models.Response
http_header = {}
http_password = None
http_proxy = None
http_session() → requests.sessions.Session
http_timeout_max_tries = 3
http_timeout_sec = 30
http_user_agent = '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 = True
https_proxy = None
setup()
ssl_client_cert = None
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: typing.Any, ttl: typing.Union[int, NoneType] = None)
redis_cache_db = 9
redis_cache_host = '127.0.0.1'
redis_cache_password = None
redis_cache_port = 6379
redis_cache_ttl = 15
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)
message_jsondict_as_string = True
reconnect_delay = 0