Database

This module provides database functions for working with Karton tasks in Artemis.

class artemis.db.Analysis(**kwargs)

Bases: Base

Represents an analysis entry in the Artemis system. An analysis is a scan of an entered domain and can consist of multiple TaskResult.

Variables:

  • id – Unique identifier for the analysis.

  • created_at – Timestamp when the analysis was created.

  • target – Target of the analysis (e.g., domain, IP).

  • tag – Tag associated with the analysis.

  • stopped – Whether the analysis has been stopped.

  • disabled_modules – Comma-separated list of disabled modules for this analysis.

class artemis.db.ColumnOrdering(column_name: str, ascending: bool)

Bases: object

class artemis.db.PaginatedResults(records_count_total: int, records_count_filtered: int, data: List[Dict[str, Any]])

Bases: object

class artemis.db.ReportGenerationTask(**kwargs)

Bases: Base

Represents a report generation task in the Artemis system. For more information on report generation, see Generating reports to be sent

Variables:

  • id – Unique identifier for the report generation task.

  • created_at – Timestamp when the report generation task was created.

  • comment – Optional comment for the task.

  • status – Status of the report generation task.

  • tag – Optional tag for the task.

  • language – Language for the report.

  • skip_previously_exported – Whether to skip previously exported reports.

  • skip_hooks – Whether to skip hooks during report generation.

  • skip_suspicious_reports – Whether to skip suspicious reports.

  • custom_template_arguments – Custom arguments for the report template.

  • output_location – Output location for the report.

  • error – Error message, if any.

  • alerts – Alerts associated with the report generation task.

class artemis.db.ReportGenerationTaskStatus(value)

Bases: str, Enum

class artemis.db.ScheduledTask(**kwargs)

Bases: Base

Represents a scheduled task in the Artemis system. The purpose of this object is solely to prevent duplication, not to perform actual task scheduling.

Variables:

  • analysis_id – Unique identifier for the analysis associated with this scheduled task.

  • deduplication_data – Hash used for deduplication of scheduled tasks.

  • task_id – Unique identifier for the underlying task.

  • created_at – Timestamp when the scheduled task was created.

class artemis.db.TSVector(*args: Any, **kwargs: Any)

Bases: TypeDecorator

impl

alias of TSVECTOR

class artemis.db.Tag(**kwargs)

Bases: Base

Represents a tag used for categorizing analyses.

Variables:

  • id – Unique identifier for the tag.

  • tag_name – Name of the tag.

  • created_at – Timestamp when the tag was created.

class artemis.db.TagArchiveRequest(**kwargs)

Bases: Base

class artemis.db.TaskFilter(value)

Bases: str, Enum

class artemis.db.TaskResult(**kwargs)

Bases: Base

Stores the result of a Karton Task execution.

Variables:

  • id – Unique identifier for the task result.

  • analysis_id – Identifier of the related analysis.

  • created_at – Timestamp when the result was created.

  • status – TaskStatus (e.g., OK, ERROR, INTERESTING).

  • tag – Tag associated with the result.

  • receiver – Receiver module for the task.

  • target_string – Target string for the result.

  • status_reason – Reason for the status.

  • headers_string – String of all headers for searching.

  • logs – Logs associated with the task.

  • task – JSON representation of the original task.

  • result – JSON representation of the result data.

  • additional_info – JSON representation of the additional data that can be saved

Classes without docstrings

class artemis.db.ColumnOrdering(column_name: str, ascending: bool)
ascending: bool
column_name: str
class artemis.db.TSVector(*args: Any, **kwargs: Any)
impl

alias of TSVECTOR

class artemis.db.PaginatedResults(records_count_total: int, records_count_filtered: int, data: List[Dict[str, Any]])
data: List[Dict[str, Any]]
records_count_filtered: int
records_count_total: int
class artemis.db.TagArchiveRequest(**kwargs)
created_at
id
tag
class artemis.db.DB
create_analysis(analysis: Task) → None

Create a new Analysis entry in the database from a given Karton Task.

Parameters:

analysis (Task) – The Karton Task object containing analysis information.

Returns:

None

create_report_generation_task(tag: str | None, comment: str | None, language: Language, skip_previously_exported: bool, skip_hooks: bool = False, skip_suspicious_reports: bool = False, custom_template_arguments: Dict[str, Any] = {}, include_only_results_since: datetime | None = None) → None
create_tag_archive_request(tag: str) → None
delete_analysis(id: str) → None
delete_report_generation_task(id: int) → None
delete_scheduled_task(scheduled_task_id: str) → int
delete_scheduled_tasks_for_analyses(analysis_ids: list[str]) → int
delete_tag_archive_request(tag: str) → None
delete_task_result(id: str) → None
delete_task_results_by_ids(ids: List[str]) → None
static dict_to_str(d: Dict[str, Any]) → str
get_analysis_by_id(analysis_id: str) → Dict[str, Any] | None

Retrieve an Analysis entry from the database by its analysis ID.

Parameters:

analysis_id (str) – The unique identifier of the analysis to retrieve. It’s Task.root_uuid.

Returns:

The analysis data as a dictionary if found, otherwise None.

Return type:

Optional[Dict[str, Any]]

get_oldest_task_results_before(time_to: datetime, max_length: int, interesting: bool, batch_size: int = 100) → List[Dict[str, Any]]
get_oldest_task_results_with_tag(tag: str, max_length: int, batch_size: int = 100) → List[Dict[str, Any]]
get_paginated_analyses(start: int, length: int, ordering: List[ColumnOrdering], *, search_query: str | None = None) → PaginatedResults
get_paginated_task_results(start: int, length: int, ordering: List[ColumnOrdering], *, search_query: str | None = None, analysis_id: str | None = None, task_filter: TaskFilter | None = None, apply_custom_filter: Callable | None = None) → PaginatedResults
get_report_generation_task(id: int) → ReportGenerationTask | None
get_tags() → List[Type[Tag]] | Any
get_task_by_id(task_id: str) → Dict[str, Any] | None
get_task_results_since(time_from: datetime, tag: str | None = None, batch_size: int = 100) → Generator[Dict[str, Any], None, None]
list_analysis() → List[Dict[str, Any]]
list_report_generation_tasks(tag_prefix: str | None = None) → List[ReportGenerationTask]
list_tag_archive_requests(min_age: datetime) → List[Dict[str, Any]]
mark_analysis_as_stopped(analysis_id: str) → None
save_report_generation_task_results(task: ReportGenerationTask, status: ReportGenerationTaskStatus, output_location: str | None = None, error: str | None = None, alerts: List[str] | None = None) → None
save_scheduled_task(task: Task) → bool

Saves a scheduled task and returns True if it didn’t exist in the database.

The purpose of this method is deduplication - making sure identical tasks aren’t run twice.

save_tag(tag_name: str | None) → None
save_task_logs(task_id: str, logs: bytes) → None
save_task_result(task: Task, *, status: TaskStatus, status_reason: str | None = None, data: Any | None = None) → None

Save the result of a Task execution to the database.

Parameters:

  • task (Task) – The Karton Task object for which the result is being saved.

  • status (TaskStatus) – The status of the task (e.g., OK, INTERESTING, etc.).

  • status_reason (Optional[str]) – Optional reason or message describing the status.

  • data (Optional[Any]) – Optional result data, can be a Pydantic model, Exception, or any serializable object.

Returns:

None

take_single_report_generation_task() → ReportGenerationTask | None
task_to_dict(task: Task) → Dict[str, Any]
class artemis.db.TestDB
delete_task_results() → None
get_single_task_result() → Dict[str, Any]