Skip to content

API Reference

thehive4py

client

TheHiveApi(url, apikey=None, username=None, password=None, organisation=None, verify=True, max_retries=DEFAULT_RETRY)

Create a client of TheHive API.

Parameters:

Name Type Description Default
url str

TheHive's url.

required
apikey Optional[str]

TheHive's apikey. It's required if username and password is not provided.

None
username Optional[str]

TheHive's username. It's required if apikey is not provided. Must be specified together with password.

None
password Optional[str]

TheHive's password. It's required if apikey is not provided. Must be specified together with username.

None
organisation Optional[str]

TheHive organisation to use in the session.

None
verify VerifyValue

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.

True
max_retries RetryValue

Either None, in which case we do not retry failed requests, or a Retry object.

DEFAULT_RETRY
Source code in thehive4py/client.py
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    url: str,
    apikey: Optional[str] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    organisation: Optional[str] = None,
    verify: VerifyValue = True,
    max_retries: RetryValue = DEFAULT_RETRY,
):
    """Create a client of TheHive API.

    Parameters:
        url: TheHive's url.
        apikey: TheHive's apikey. It's required if `username` and `password`
            is not provided.
        username: TheHive's username. It's required if `apikey` is not provided.
            Must be specified together with `password`.
        password: TheHive's password. It's required if `apikey` is not provided.
            Must be specified together with `username`.
        organisation: TheHive organisation to use in the session.
        verify: 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.
        max_retries: Either `None`, in which case we do not retry failed requests,
            or a `Retry` object.

    """
    self.session = TheHiveSession(
        url=url,
        apikey=apikey,
        username=username,
        password=password,
        verify=verify,
        max_retries=max_retries,
    )
    self.session_organisation = organisation

    # case management endpoints
    self.alert = AlertEndpoint(self.session)
    self.case = CaseEndpoint(self.session)
    self.case_template = CaseTemplateEndpoint(self.session)
    self.comment = CommentEndpoint(self.session)
    self.observable = ObservableEndpoint(self.session)
    self.procedure = ProcedureEndpoint(self.session)
    self.task = TaskEndpoint(self.session)
    self.task_log = TaskLogEndpoint(self.session)
    self.timeline = TimelineEndpoint(self.session)

    # user management endpoints
    self.user = UserEndpoint(self.session)
    self.organisation = OrganisationEndpoint(self.session)
    self.profile = ProfileEndpoint(self.session)

    # entity endpoints
    self.custom_field = CustomFieldEndpoint(self.session)
    self.observable_type = ObservableTypeEndpoint(self.session)

    # connector endpoints
    self.cortex = CortexEndpoint(self.session)

    # standard endpoints
    self.query = QueryEndpoint(self.session)
session = TheHiveSession(url=url, apikey=apikey, username=username, password=password, verify=verify, max_retries=max_retries) instance-attribute
alert = AlertEndpoint(self.session) instance-attribute
case = CaseEndpoint(self.session) instance-attribute
case_template = CaseTemplateEndpoint(self.session) instance-attribute
comment = CommentEndpoint(self.session) instance-attribute
observable = ObservableEndpoint(self.session) instance-attribute
procedure = ProcedureEndpoint(self.session) instance-attribute
task = TaskEndpoint(self.session) instance-attribute
task_log = TaskLogEndpoint(self.session) instance-attribute
timeline = TimelineEndpoint(self.session) instance-attribute
user = UserEndpoint(self.session) instance-attribute
organisation = OrganisationEndpoint(self.session) instance-attribute
profile = ProfileEndpoint(self.session) instance-attribute
custom_field = CustomFieldEndpoint(self.session) instance-attribute
observable_type = ObservableTypeEndpoint(self.session) instance-attribute
cortex = CortexEndpoint(self.session) instance-attribute
query = QueryEndpoint(self.session) instance-attribute
session_organisation property writable

session

DEFAULT_RETRY = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504], allowed_methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE'], raise_on_status=False) module-attribute

RetryValue = Union[Retry, int, None] module-attribute

VerifyValue = Union[bool, str] module-attribute

_SessionJSONEncoder

Bases: JSONEncoder

Custom JSON encoder class for TheHive session.

default(o)
Source code in thehive4py/session.py
30
31
32
33
def default(self, o: Any):
    if isinstance(o, UserDict):
        return o.data
    return super().default(o)

TheHiveSession(url, apikey=None, username=None, password=None, verify=True, max_retries=DEFAULT_RETRY)

Bases: Session

Source code in thehive4py/session.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    url: str,
    apikey: Optional[str] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    verify: VerifyValue = True,
    max_retries: RetryValue = DEFAULT_RETRY,
):
    super().__init__()
    self.hive_url = self._sanitize_hive_url(url)
    self.verify = verify
    self.headers["User-Agent"] = f"thehive4py/{__version__}"
    self._set_retries(max_retries=max_retries)

    if username and password:
        self.headers["Authorization"] = requests.auth._basic_auth_str(
            username, password
        )
    elif apikey:
        self.headers["Authorization"] = f"Bearer {apikey}"
    else:
        raise TheHiveError(
            "Either apikey or the username/password combination must be provided!"
        )
hive_url = self._sanitize_hive_url(url) instance-attribute
verify = verify instance-attribute
_set_retries(max_retries)

Configure the session to retry.

Source code in thehive4py/session.py
63
64
65
66
67
def _set_retries(self, max_retries: RetryValue):
    """Configure the session to retry."""
    retry_adapter = requests.adapters.HTTPAdapter(max_retries=max_retries)
    self.mount("http://", retry_adapter)
    self.mount("https://", retry_adapter)
_sanitize_hive_url(hive_url)

Sanitize the base url for the client.

Source code in thehive4py/session.py
69
70
71
72
73
def _sanitize_hive_url(self, hive_url: str) -> str:
    """Sanitize the base url for the client."""
    if hive_url.endswith("/"):
        return hive_url[:-1]
    return hive_url
make_request(method, path, params=None, data=None, json=None, files=None, download_path=None)
Source code in thehive4py/session.py
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def make_request(
    self,
    method: str,
    path: str,
    params=None,
    data=None,
    json=None,
    files=None,
    download_path: Union[str, PathLike, None] = None,
) -> Any:
    endpoint_url = f"{self.hive_url}{path}"

    headers = {**self.headers}
    if json:
        data = jsonlib.dumps(json, cls=_SessionJSONEncoder)
        headers = {**headers, "Content-Type": "application/json"}

    response = self.request(
        method,
        url=endpoint_url,
        params=params,
        data=data,
        files=files,
        headers=headers,
        verify=self.verify,
        stream=bool(download_path),
    )

    return self._process_response(response, download_path=download_path)
_process_response(response, download_path=None)
Source code in thehive4py/session.py
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def _process_response(
    self,
    response: requests.Response,
    download_path: Union[str, PathLike, None] = None,
):
    if response.ok:
        if download_path is None:
            return self._process_text_response(response)
        else:
            self._process_stream_response(
                response=response, download_path=download_path
            )

    if not response.ok:
        self._process_error_response(response=response)
_process_text_response(response)
Source code in thehive4py/session.py
121
122
123
124
125
126
127
128
129
def _process_text_response(self, response: requests.Response):
    try:
        json_data = response.json()
    except requests.exceptions.JSONDecodeError:
        json_data = None

    if json_data is None:
        return response.text
    return json_data
_process_stream_response(response, download_path)
Source code in thehive4py/session.py
131
132
133
134
135
136
def _process_stream_response(
    self, response: requests.Response, download_path: Union[str, PathLike]
):
    with open(download_path, "wb") as download_fp:
        for chunk in response.iter_content(chunk_size=4096):
            download_fp.write(chunk)
_process_error_response(response)
Source code in thehive4py/session.py
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def _process_error_response(self, response: requests.Response):
    try:
        json_data = response.json()
    except requests.exceptions.JSONDecodeError:
        json_data = None

    if isinstance(json_data, dict) and all(
        [
            "type" in json_data,
            "message" in json_data,
        ]
    ):
        error_text = f"{json_data['type']} - {json_data['message']}"
    else:
        error_text = response.text
    raise TheHiveError(message=error_text, response=response)

endpoints

_base

EndpointBase(session)
Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
_session = session instance-attribute
_fileinfo_from_filepath(filepath)
Source code in thehive4py/endpoints/_base.py
18
19
20
21
22
23
def _fileinfo_from_filepath(self, filepath: str) -> tuple:
    filename = os.path.split(filepath)[1]
    mimetype = mimetypes.guess_type(filepath)[0]
    filestream = open(filepath, "rb")

    return (filename, filestream, mimetype)
_build_observable_kwargs(observable, observable_path=None)
Source code in thehive4py/endpoints/_base.py
25
26
27
28
29
30
31
32
33
34
35
36
def _build_observable_kwargs(
    self, observable: InputObservable, observable_path: Optional[str] = None
) -> dict:
    if observable_path:
        kwargs = {
            "data": {"_json": json.dumps(observable)},
            "files": {"attachment": self._fileinfo_from_filepath(observable_path)},
        }
    else:
        kwargs = {"json": observable}

    return kwargs
_build_subquery(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/_base.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def _build_subquery(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> QueryExpr:
    subquery: QueryExpr = []
    if filters:
        subquery = [*subquery, {"_name": "filter", **filters}]
    if sortby:
        subquery = [*subquery, {"_name": "sort", **sortby}]
    if paginate:
        subquery = [*subquery, {"_name": "page", **paginate}]

    return subquery

alert

AlertEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(alert, attachment_map=None)

Create an alert.

Parameters:

Name Type Description Default
alert InputAlert

The body of the alert.

required
attachment_map Optional[Dict[str, str]]

An optional mapping of observable attachment keys and paths.

None

Returns:

Type Description
OutputAlert

The created alert.

Source code in thehive4py/endpoints/alert.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def create(
    self, alert: InputAlert, attachment_map: Optional[Dict[str, str]] = None
) -> OutputAlert:
    """Create an alert.

    Args:
        alert: The body of the alert.
        attachment_map: An optional mapping of observable attachment keys and paths.

    Returns:
        The created alert.
    """
    if attachment_map:
        files: Dict[str, Any] = {
            key: self._fileinfo_from_filepath(path)
            for key, path in attachment_map.items()
        }
        files["_json"] = jsonlib.dumps(alert)
        kwargs: dict = {"files": files}
    else:
        kwargs = {"json": alert}
    return self._session.make_request("POST", path="/api/v1/alert", **kwargs)
get(alert_id)

Get an alert by id.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required

Returns:

Type Description
OutputAlert

The alert specified by the id.

Source code in thehive4py/endpoints/alert.py
47
48
49
50
51
52
53
54
55
56
57
def get(self, alert_id: str) -> OutputAlert:
    """Get an alert by id.

    Args:
        alert_id: The id of the alert.

    Returns:
        The alert specified by the id.
    """

    return self._session.make_request("GET", path=f"/api/v1/alert/{alert_id}")
update(alert_id, fields)

Update an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
fields InputUpdateAlert

The fields of the alert to update.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
59
60
61
62
63
64
65
66
67
68
69
70
71
def update(self, alert_id: str, fields: InputUpdateAlert) -> None:
    """Update an alert.

    Args:
        alert_id: The id of the alert.
        fields: The fields of the alert to update.

    Returns:
        N/A
    """
    return self._session.make_request(
        "PATCH", path=f"/api/v1/alert/{alert_id}", json=fields
    )
delete(alert_id)

Delete an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
73
74
75
76
77
78
79
80
81
82
def delete(self, alert_id: str) -> None:
    """Delete an alert.

    Args:
        alert_id: The id of the alert.

    Returns:
        N/A
    """
    return self._session.make_request("DELETE", path=f"/api/v1/alert/{alert_id}")
bulk_update(fields)

Update multiple alerts with the same values.

Parameters:

Name Type Description Default
fields InputBulkUpdateAlert

The ids and the fields of the alerts to update.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
84
85
86
87
88
89
90
91
92
93
94
95
def bulk_update(self, fields: InputBulkUpdateAlert) -> None:
    """Update multiple alerts with the same values.

    Args:
        fields: The ids and the fields of the alerts to update.

    Returns:
        N/A
    """
    return self._session.make_request(
        "PATCH", path="/api/v1/alert/_bulk", json=fields
    )
bulk_delete(ids)

Delete multiple alerts.

Parameters:

Name Type Description Default
ids List[str]

The ids of the alerts to delete.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
 97
 98
 99
100
101
102
103
104
105
106
107
108
def bulk_delete(self, ids: List[str]) -> None:
    """Delete multiple alerts.

    Args:
        ids: The ids of the alerts to delete.

    Returns:
        N/A
    """
    return self._session.make_request(
        "POST", path="/api/v1/alert/delete/_bulk", json={"ids": ids}
    )
follow(alert_id)

Follow an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
110
111
112
113
114
115
116
117
118
119
def follow(self, alert_id: str) -> None:
    """Follow an alert.

    Args:
        alert_id: The id of the alert.

    Returns:
        N/A
    """
    self._session.make_request("POST", path=f"/api/v1/alert/{alert_id}/follow")
unfollow(alert_id)

Unfollow an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
121
122
123
124
125
126
127
128
129
130
def unfollow(self, alert_id: str) -> None:
    """Unfollow an alert.

    Args:
        alert_id: The id of the alert.

    Returns:
        N/A
    """
    self._session.make_request("POST", path=f"/api/v1/alert/{alert_id}/unfollow")
promote_to_case(alert_id, fields={})

Promote an alert into a case.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
fields InputPromoteAlert

Override for the fields of the case created from the alert.

{}

Returns:

Type Description
OutputCase

The case from the promoted alert.

Source code in thehive4py/endpoints/alert.py
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def promote_to_case(
    self, alert_id: str, fields: InputPromoteAlert = {}
) -> OutputCase:
    """Promote an alert into a case.

    Args:
        alert_id: The id of the alert.
        fields: Override for the fields of the case created from the alert.

    Returns:
        The case from the promoted alert.
    """
    return self._session.make_request(
        "POST",
        path=f"/api/v1/alert/{alert_id}/case",
        json=fields,
    )
create_observable(alert_id, observable, observable_path=None)

Create an observable in an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
observable InputObservable

The fields of the observable to create.

required
observable_path Optional[str]

Optional path in case of a file based observable.

None

Returns:

Type Description
List[OutputObservable]

The created alert observables.

Source code in thehive4py/endpoints/alert.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def create_observable(
    self,
    alert_id: str,
    observable: InputObservable,
    observable_path: Optional[str] = None,
) -> List[OutputObservable]:
    """Create an observable in an alert.

    Args:
        alert_id: The id of the alert.
        observable: The fields of the observable to create.
        observable_path: Optional path in case of a file based observable.

    Returns:
        The created alert observables.
    """

    kwargs = self._build_observable_kwargs(
        observable=observable, observable_path=observable_path
    )
    return self._session.make_request(
        "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
    )
add_attachment(alert_id, attachment_paths)

Create an observable in an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
attachment_paths List[str]

List of paths to the attachments to create.

required

Returns:

Type Description
List[OutputAttachment]

The created alert attachments.

Source code in thehive4py/endpoints/alert.py
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def add_attachment(
    self, alert_id: str, attachment_paths: List[str]
) -> List[OutputAttachment]:
    """Create an observable in an alert.

    Args:
        alert_id: The id of the alert.
        attachment_paths: List of paths to the attachments to create.

    Returns:
        The created alert attachments.
    """
    files = [
        ("attachments", self._fileinfo_from_filepath(attachment_path))
        for attachment_path in attachment_paths
    ]
    return self._session.make_request(
        "POST", f"/api/v1/alert/{alert_id}/attachments", files=files
    )["attachments"]
download_attachment(alert_id, attachment_id, attachment_path)

Download an alert attachment.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
attachment_id str

The id of the alert attachment.

required
attachment_path str

The local path to download the attachment to.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def download_attachment(
    self, alert_id: str, attachment_id: str, attachment_path: str
) -> None:
    """Download an alert attachment.

    Args:
        alert_id: The id of the alert.
        attachment_id: The id of the alert attachment.
        attachment_path: The local path to download the attachment to.

    Returns:
        N/A
    """
    return self._session.make_request(
        "GET",
        path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}/download",
        download_path=attachment_path,
    )
delete_attachment(alert_id, attachment_id)

Delete an alert attachment.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
attachment_id str

The id of the alert attachment.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/alert.py
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def delete_attachment(self, alert_id: str, attachment_id: str) -> None:
    """Delete an alert attachment.

    Args:
        alert_id: The id of the alert.
        attachment_id: The id of the alert attachment.

    Returns:
        N/A
    """

    return self._session.make_request(
        "DELETE", path=f"/api/v1/alert/{alert_id}/attachment/{attachment_id}"
    )
merge_into_case(alert_id, case_id)

Merge an alert into an existing case.

Parameters:

Name Type Description Default
alert_id str

The id of the alert to merge.

required
case_id str

The id of the case to merge the alert into.

required

Returns:

Type Description
OutputCase

The case into which the alert was merged.

Source code in thehive4py/endpoints/alert.py
228
229
230
231
232
233
234
235
236
237
238
239
240
def merge_into_case(self, alert_id: str, case_id: str) -> OutputCase:
    """Merge an alert into an existing case.

    Args:
        alert_id: The id of the alert to merge.
        case_id: The id of the case to merge the alert into.

    Returns:
        The case into which the alert was merged.
    """
    return self._session.make_request(
        "POST", path=f"/api/v1/alert/{alert_id}/merge/{case_id}"
    )
bulk_merge_into_case(case_id, alert_ids)

Merge an alert into an existing case.

Parameters:

Name Type Description Default
case_id str

The id of the case to merge the alerts into.

required
alert_ids List[str]

The list of alert ids to merge.

required

Returns:

Type Description
OutputCase

The case into which the alerts were merged.

Source code in thehive4py/endpoints/alert.py
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
def bulk_merge_into_case(self, case_id: str, alert_ids: List[str]) -> OutputCase:
    """Merge an alert into an existing case.

    Args:
        case_id: The id of the case to merge the alerts into.
        alert_ids: The list of alert ids to merge.

    Returns:
        The case into which the alerts were merged.
    """
    return self._session.make_request(
        "POST",
        path="/api/v1/alert/merge/_bulk",
        json={"caseId": case_id, "alertIds": alert_ids},
    )
find(filters=None, sortby=None, paginate=None)

Find multiple alerts.

Parameters:

Name Type Description Default
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputAlert]

The list of alerts matched by the query or an empty list.

Source code in thehive4py/endpoints/alert.py
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputAlert]:
    """Find multiple alerts.

    Args:
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of alerts matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "listAlert"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "alerts"},
        json={"query": query},
    )
count(filters=None)

Count alerts.

Parameters:

Name Type Description Default
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None

Returns:

Type Description
int

The count of alerts matched by the query.

Source code in thehive4py/endpoints/alert.py
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def count(self, filters: Optional[FilterExpr] = None) -> int:
    """Count alerts.

    Args:
        filters: The filter expressions to apply in the query.

    Returns:
        The count of alerts matched by the query.
    """

    query: QueryExpr = [
        {"_name": "listAlert"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "alerts.count"},
        json={"query": query},
    )
find_observables(alert_id, filters=None, sortby=None, paginate=None)

Find observable related to an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputObservable]

The list of alert observables matched by the query or an empty list.

Source code in thehive4py/endpoints/alert.py
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def find_observables(
    self,
    alert_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputObservable]:
    """Find observable related to an alert.

    Args:
        alert_id: The id of the alert.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of alert observables matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getAlert", "idOrName": alert_id},
        {"_name": "observables"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "alert-observables"},
        json={"query": query},
    )
find_comments(alert_id, filters=None, sortby=None, paginate=None)

Find comments related to an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputComment]

The list of alert comments matched by the query or an empty list.

Source code in thehive4py/endpoints/alert.py
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
def find_comments(
    self,
    alert_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputComment]:
    """Find comments related to an alert.

    Args:
        alert_id: The id of the alert.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of alert comments matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getAlert", "idOrName": alert_id},
        {"_name": "comments"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "alert-comments"},
        json={"query": query},
    )
create_procedure(alert_id, procedure)

Create an alert procedure.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
procedure InputProcedure

The fields of the procedure to create.

required

Returns:

Type Description
OutputProcedure

The created alert procedure.

Source code in thehive4py/endpoints/alert.py
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def create_procedure(
    self, alert_id: str, procedure: InputProcedure
) -> OutputProcedure:
    """Create an alert procedure.

    Args:
        alert_id: The id of the alert.
        procedure: The fields of the procedure to create.

    Returns:
        The created alert procedure.
    """
    return self._session.make_request(
        "POST", path=f"/api/v1/alert/{alert_id}/procedure", json=procedure
    )
find_procedures(alert_id, filters=None, sortby=None, paginate=None)

Find procedures related to an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputProcedure]

The list of alert procedures matched by the query or an empty list.

Source code in thehive4py/endpoints/alert.py
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
def find_procedures(
    self,
    alert_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProcedure]:
    """Find procedures related to an alert.

    Args:
        alert_id: The id of the alert.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of alert procedures matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getAlert", "idOrName": alert_id},
        {"_name": "procedures"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "alert-procedures"},
        json={"query": query},
    )
find_attachments(alert_id, filters=None, sortby=None, paginate=None)

Find attachments related to an alert.

Parameters:

Name Type Description Default
alert_id str

The id of the alert.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputAttachment]

The list of alert attachments matched by the query or an empty list.

Source code in thehive4py/endpoints/alert.py
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
def find_attachments(
    self,
    alert_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputAttachment]:
    """Find attachments related to an alert.

    Args:
        alert_id: The id of the alert.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of alert attachments matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getAlert", "idOrName": alert_id},
        {"_name": "attachments"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "alert-attachments"},
        json={"query": query},
    )

case

CaseId = Union[str, int] module-attribute
CaseEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(case)

Create a case.

Parameters:

Name Type Description Default
case InputCase

The body of the case.

required

Returns:

Type Description
OutputCase

The created case.

Source code in thehive4py/endpoints/case.py
35
36
37
38
39
40
41
42
43
44
def create(self, case: InputCase) -> OutputCase:
    """Create a case.

    Args:
        case: The body of the case.

    Returns:
        The created case.
    """
    return self._session.make_request("POST", path="/api/v1/case", json=case)
get(case_id)

Get a case by id.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required

Returns:

Type Description
OutputCase

The case specified by the id.

Source code in thehive4py/endpoints/case.py
46
47
48
49
50
51
52
53
54
55
def get(self, case_id: CaseId) -> OutputCase:
    """Get a case by id.

    Args:
        case_id: The id of the case.

    Returns:
        The case specified by the id.
    """
    return self._session.make_request("GET", path=f"/api/v1/case/{case_id}")
delete(case_id)

Delete a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
57
58
59
60
61
62
63
64
65
66
def delete(self, case_id: CaseId) -> None:
    """Delete a case.

    Args:
        case_id: The id of the case.

    Returns:
        N/A
    """
    self._session.make_request("DELETE", path=f"/api/v1/case/{case_id}")
update(case_id, fields={}, **kwargs)

Update a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
fields Optional[InputUpdateCase]

The fields of the case to update.

{}

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def update(
    self, case_id: CaseId, fields: Optional[InputUpdateCase] = {}, **kwargs
) -> None:
    """Update a case.

    Args:
        case_id: The id of the case.
        fields: The fields of the case to update.

    Returns:
        N/A
    """

    if not fields:
        if "case" not in kwargs:
            raise TheHiveError(
                f"Unrecognized keyword arguments: {list(kwargs.keys())}. "
                "Please use the `fields` argument to supply case update values."
            )
        warnings.warn(
            message="The `case` argument has been deprecated to follow the same "
            "convention like other update methods. Please use the `fields` "
            "argument to prevent breaking changes in the future.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        fields = kwargs["case"]

    return self._session.make_request(
        "PATCH", path=f"/api/v1/case/{case_id}", json=fields
    )
bulk_update(fields)

Update multiple cases with the same values.

Parameters:

Name Type Description Default
fields InputBulkUpdateCase

The ids and the fields of the cases to update.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
100
101
102
103
104
105
106
107
108
109
110
111
def bulk_update(self, fields: InputBulkUpdateCase) -> None:
    """Update multiple cases with the same values.

    Args:
        fields: The ids and the fields of the cases to update.

    Returns:
        N/A
    """
    return self._session.make_request(
        "PATCH", path="/api/v1/case/_bulk", json=fields
    )
merge(case_ids)

Merge multiple cases into one final case.

Parameters:

Name Type Description Default
case_ids Sequence[CaseId]

The ids of the cases to merge.

required

Returns:

Type Description
OutputCase

The merged case.

Source code in thehive4py/endpoints/case.py
113
114
115
116
117
118
119
120
121
122
123
124
125
def merge(self, case_ids: Sequence[CaseId]) -> OutputCase:
    """Merge multiple cases into one final case.

    Args:
        case_ids: The ids of the cases to merge.

    Returns:
        The merged case.
    """
    case_id_subpath = ",".join([str(case_id) for case_id in case_ids])
    return self._session.make_request(
        "POST", path=f"/api/v1/case/_merge/{case_id_subpath}"
    )

Unlink an alert from a case.

Parameters:

Name Type Description Default
case_id str

The id of the case to unlink the alert from.

required
alert_id str

The id of the alert to unlink.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
127
128
129
130
131
132
133
134
135
136
137
138
139
def unlink_alert(self, case_id: str, alert_id: str) -> None:
    """Unlink an alert from a case.

    Args:
        case_id: The id of the case to unlink the alert from.
        alert_id: The id of the alert to unlink.

    Returns:
        N/A
    """
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/{case_id}/alert/{alert_id}"
    )
merge_similar_observables(case_id)

Merge similar observables of a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case to merge similar observables for.

required

Returns:

Type Description
dict

The metadata of the observable merge operation.

Source code in thehive4py/endpoints/case.py
141
142
143
144
145
146
147
148
149
150
151
152
153
def merge_similar_observables(self, case_id: CaseId) -> dict:
    """Merge similar observables of a case.

    Args:
        case_id: The id of the case to merge similar observables for.

    Returns:
        The metadata of the observable merge operation.
    """
    # TODO: add better return value type hint
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/observable/_merge"
    )
get_linked_cases(case_id)

Get other cases linked to a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case to get linked cases for.

required

Returns:

Type Description
List[OutputCase]

The list of linked cases.

Source code in thehive4py/endpoints/case.py
155
156
157
158
159
160
161
162
163
164
def get_linked_cases(self, case_id: CaseId) -> List[OutputCase]:
    """Get other cases linked to a case.

    Args:
        case_id: The id of the case to get linked cases for.

    Returns:
        The list of linked cases.
    """
    return self._session.make_request("GET", path=f"/api/v1/case/{case_id}/links")
delete_custom_field(custom_field_id)

Delete a custom field from a case.

Parameters:

Name Type Description Default
custom_field_id str

The id of the specific custom field to delete from a case.

required
Retruns

N/A

Source code in thehive4py/endpoints/case.py
166
167
168
169
170
171
172
173
174
175
176
177
def delete_custom_field(self, custom_field_id: str) -> None:
    """Delete a custom field from a case.

    Args:
        custom_field_id: The id of the specific custom field to delete from a case.

    Retruns:
        N/A
    """
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/customField/{custom_field_id}"
    )
import_from_file(import_case, import_path)

Import a case from a .thar archive file.

Parameters:

Name Type Description Default
import_case InputImportCase

The metadata of the case import.

required
import_path str

The filepath to the .thar archive.

required

Returns:

Type Description
dict

The metadata of the case import operation.

Source code in thehive4py/endpoints/case.py
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def import_from_file(self, import_case: InputImportCase, import_path: str) -> dict:
    """Import a case from a .thar archive file.

    Args:
        import_case: The metadata of the case import.
        import_path: The filepath to the .thar archive.

    Returns:
        The metadata of the case import operation.
    """
    # TODO: add better return type hints
    return self._session.make_request(
        "POST",
        path="/api/v1/case/import",
        data={"_json": jsonlib.dumps(import_case)},
        files={"file": self._fileinfo_from_filepath(import_path)},
    )
export_to_file(case_id, password, export_path)

Export a case to a .thar archive file.

The file can be used to import the case in an other TheHive instance

Parameters:

Name Type Description Default
case_id CaseId

The id of the case to export.

required
password str

The password to encrypt the .thar file with.

required
export_path str

The filepath to save the case export to.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def export_to_file(self, case_id: CaseId, password: str, export_path: str) -> None:
    """Export a case to a .thar archive file.

    The file can be used to import the case in an other TheHive instance

    Args:
        case_id: The id of the case to export.
        password: The password to encrypt the .thar file with.
        export_path: The filepath to save the case export to.

    Returns:
        N/A
    """
    return self._session.make_request(
        "GET",
        path=f"/api/v1/case/{case_id}/export",
        params={"password": password},
        download_path=export_path,
    )
get_timeline(case_id)

Get the timeline of a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case with the timeline.

required

Returns:

Type Description
OutputTimeline

The case timeline.

Source code in thehive4py/endpoints/case.py
217
218
219
220
221
222
223
224
225
226
def get_timeline(self, case_id: CaseId) -> OutputTimeline:
    """Get the timeline of a case.

    Args:
        case_id: The id of the case with the timeline.

    Returns:
        The case timeline.
    """
    return self._session.make_request("GET", f"/api/v1/case/{case_id}/timeline")
apply_case_template(fields)

Retroactively apply a case template on a case.

Parameters:

Name Type Description Default
fields InputApplyCaseTemplate

The metadata of the case template apply operation.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
228
229
230
231
232
233
234
235
236
237
238
239
def apply_case_template(self, fields: InputApplyCaseTemplate) -> None:
    """Retroactively apply a case template on a case.

    Args:
        fields: The metadata of the case template apply operation.

    Returns:
        N/A
    """
    return self._session.make_request(
        "POST", "/api/v1/case/_bulk/caseTemplate", json=fields
    )
add_attachment(case_id, attachment_paths)

Create an attachment in a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
attachment_paths List[str]

List of paths to the attachments to create.

required

Returns:

Type Description
List[OutputAttachment]

The created case attachments.

Source code in thehive4py/endpoints/case.py
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def add_attachment(
    self, case_id: CaseId, attachment_paths: List[str]
) -> List[OutputAttachment]:
    """Create an attachment in a case.

    Args:
        case_id: The id of the case.
        attachment_paths: List of paths to the attachments to create.

    Returns:
        The created case attachments.
    """

    files = [
        ("attachments", self._fileinfo_from_filepath(attachment_path))
        for attachment_path in attachment_paths
    ]
    return self._session.make_request(
        "POST", f"/api/v1/case/{case_id}/attachments", files=files
    )["attachments"]
download_attachment(case_id, attachment_id, attachment_path)

Download a case attachment.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
attachment_id str

The id of the case attachment.

required
attachment_path str

The local path to download the attachment to.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def download_attachment(
    self, case_id: CaseId, attachment_id: str, attachment_path: str
) -> None:
    """Download a case attachment.

    Args:
        case_id: The id of the case.
        attachment_id: The id of the case attachment.
        attachment_path: The local path to download the attachment to.

    Returns:
        N/A
    """
    return self._session.make_request(
        "GET",
        path=f"/api/v1/case/{case_id}/attachment/{attachment_id}/download",
        download_path=attachment_path,
    )
delete_attachment(case_id, attachment_id)

Delete a case attachment.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
attachment_id str

The id of the case attachment.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
281
282
283
284
285
286
287
288
289
290
291
292
293
def delete_attachment(self, case_id: CaseId, attachment_id: str) -> None:
    """Delete a case attachment.

    Args:
        case_id: The id of the case.
        attachment_id: The id of the case attachment.

    Returns:
        N/A
    """
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/{case_id}/attachment/{attachment_id}"
    )
list_shares(case_id)

List all organisation shares of a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required

Returns:

Type Description
List[OutputShare]

The list of organisation shares of the case.

Source code in thehive4py/endpoints/case.py
295
296
297
298
299
300
301
302
303
304
def list_shares(self, case_id: CaseId) -> List[OutputShare]:
    """List all organisation shares of a case.

    Args:
        case_id: The id of the case.

    Returns:
        The list of organisation shares of the case.
    """
    return self._session.make_request("GET", path=f"/api/v1/case/{case_id}/shares")
share(case_id, shares)

Share the case with other organisations.

For each organisation, you can define a profile (level of access) that the org will receive. This request will only create new shares and will not update or delete existing shares.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
shares List[InputShare]

The list of organisational share rules.

required

Returns:

Type Description
List[OutputShare]

The list of organisation shares of the case.

Source code in thehive4py/endpoints/case.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def share(self, case_id: CaseId, shares: List[InputShare]) -> List[OutputShare]:
    """Share the case with other organisations.

    For each organisation, you can define a profile (level of access) that the org
    will receive. This request will only create new shares and will not update or
    delete existing shares.

    Args:
        case_id: The id of the case.
        shares: The list of organisational share rules.

    Returns:
        The list of organisation shares of the case.
    """
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/shares", json={"shares": shares}
    )
unshare(case_id, organisation_ids)

Unshare a case from other organisations.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
organisation_ids List[str]

The ids of the organisations to unshare from.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def unshare(self, case_id: CaseId, organisation_ids: List[str]) -> None:
    """Unshare a case from other organisations.

    Args:
        case_id: The id of the case.
        organisation_ids: The ids of the organisations to unshare from.

    Returns:
        N/A
    """
    return self._session.make_request(
        "DELETE",
        path=f"/api/v1/case/{case_id}/shares",
        json={"organisations": organisation_ids},
    )
set_share(case_id, shares)

Set the share for a case with other organisations.

For each organisation, you can define a profile (level of access) that the org will receive. This request can delete and update already existing shares.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
shares List[InputShare]

The list of organisational share rules.

required

Returns:

Type Description
List[OutputShare]

The list of organisation shares of the case.

Source code in thehive4py/endpoints/case.py
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
def set_share(self, case_id: CaseId, shares: List[InputShare]) -> List[OutputShare]:
    """Set the share for a case with other organisations.

    For each organisation, you can define a profile (level of access) that the org
    will receive. This request can delete and update already existing shares.

    Args:
        case_id: The id of the case.
        shares: The list of organisational share rules.

    Returns:
        The list of organisation shares of the case.
    """
    return self._session.make_request(
        "PUT", path=f"/api/v1/case/{case_id}/shares", json={"shares": shares}
    )
remove_share(share_id)

Remove a specific organisation share from a case.

Parameters:

Name Type Description Default
share_id str

The id of the share to remove.

required

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
357
358
359
360
361
362
363
364
365
366
367
368
def remove_share(self, share_id: str) -> None:
    """Remove a specific organisation share from a case.

    Args:
        share_id: The id of the share to remove.

    Returns:
        N/A
    """
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/share/{share_id}"
    )
update_share(share_id, profile)
Source code in thehive4py/endpoints/case.py
370
371
372
373
def update_share(self, share_id: str, profile: str) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/case/share/{share_id}", json={"profile": profile}
    )
find(filters=None, sortby=None, paginate=None)

Find multiple cases.

Parameters:

Name Type Description Default
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputCase]

The list of cases matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputCase]:
    """Find multiple cases.

    Args:
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of cases matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "listCase"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "cases"},
        json={"query": query},
    )
count(filters=None)

Count cases.

Parameters:

Name Type Description Default
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None

Returns:

Type Description
int

The count of cases matched by the query.

Source code in thehive4py/endpoints/case.py
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def count(self, filters: Optional[FilterExpr] = None) -> int:
    """Count cases.

    Args:
        filters: The filter expressions to apply in the query.

    Returns:
        The count of cases matched by the query.
    """
    query: QueryExpr = [
        {"_name": "listCase"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "cases.count"},
        json={"query": query},
    )
create_task(case_id, task)

Create a case task.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
task InputTask

The fields of the task to create.

required

Returns:

Type Description
OutputTask

The created case task.

Source code in thehive4py/endpoints/case.py
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
def create_task(self, case_id: CaseId, task: InputTask) -> OutputTask:
    """Create a case task.

    Args:
        case_id: The id of the case.
        task: The fields of the task to create.

    Returns:
        The created case task.
    """
    return self._session.make_request(
        "POST",
        path=f"/api/v1/case/{case_id}/task",
        json=task,
    )
find_tasks(case_id, filters=None, sortby=None, paginate=None)

Find tasks related to a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputTask]

The list of case tasks matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
def find_tasks(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputTask]:
    """Find tasks related to a case.

    Args:
        case_id: The id of the case.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of case tasks matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getCase", "idOrName": case_id},
        {"_name": "tasks"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-tasks"},
        json={"query": query},
    )
create_observable(case_id, observable, observable_path=None)

Create an observable in an case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
observable InputObservable

The fields of the observable to create.

required
observable_path Optional[str]

Optional path in case of a file based observable.

None

Returns:

Type Description
List[OutputObservable]

The created case observables.

Source code in thehive4py/endpoints/case.py
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
def create_observable(
    self,
    case_id: CaseId,
    observable: InputObservable,
    observable_path: Optional[str] = None,
) -> List[OutputObservable]:
    """Create an observable in an case.

    Args:
        case_id: The id of the case.
        observable: The fields of the observable to create.
        observable_path: Optional path in case of a file based observable.

    Returns:
        The created case observables.
    """
    kwargs = self._build_observable_kwargs(
        observable=observable, observable_path=observable_path
    )
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/observable", **kwargs
    )
find_observables(case_id, filters=None, sortby=None, paginate=None)

Find observables related to a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputObservable]

The list of case observables matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
def find_observables(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputObservable]:
    """Find observables related to a case.

    Args:
        case_id: The id of the case.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of case observables matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getCase", "idOrName": case_id},
        {"_name": "observables"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-observables"},
        json={"query": query},
    )
create_procedure(case_id, procedure)

Create a case procedure.

Parameters:

Name Type Description Default
case_id str

The id of the case.

required
procedure InputProcedure

The fields of the procedure to create.

required

Returns:

Type Description
OutputProcedure

The created case procedure.

Source code in thehive4py/endpoints/case.py
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
def create_procedure(
    self, case_id: str, procedure: InputProcedure
) -> OutputProcedure:
    """Create a case procedure.

    Args:
        case_id: The id of the case.
        procedure: The fields of the procedure to create.

    Returns:
        The created case procedure.
    """
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/procedure", json=procedure
    )
find_procedures(case_id, filters=None, sortby=None, paginate=None)

Find procedures related to a case.

Parameters:

Name Type Description Default
case_id str

The id of the case.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputProcedure]

The list of case procedures matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
def find_procedures(
    self,
    case_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProcedure]:
    """Find procedures related to a case.

    Args:
        case_id: The id of the case.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of case procedures matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getCase", "idOrName": case_id},
        {"_name": "procedures"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-procedures"},
        json={"query": query},
    )
create_page(case_id, page)

Create a page in a case.

Parameters:

Name Type Description Default
case_id str

The id of the case.

required
page InputCasePage

The fields of the page to create.

required

Returns:

Type Description
OutputCasePage

The created case page.

Source code in thehive4py/endpoints/case.py
572
573
574
575
576
577
578
579
580
581
582
583
584
def create_page(self, case_id: str, page: InputCasePage) -> OutputCasePage:
    """Create a page in a case.

    Args:
        case_id: The id of the case.
        page: The fields of the page to create.

    Returns:
        The created case page.
    """
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/page", json=page
    )
delete_page(case_id, page_id)

Delete a page from a case.

Parameters:

Name Type Description Default
case_id str

The id of the case.

required
page_id str

The id of the page to delete.

required
Retruns

N/A

Source code in thehive4py/endpoints/case.py
586
587
588
589
590
591
592
593
594
595
596
597
598
def delete_page(self, case_id: str, page_id: str) -> None:
    """Delete a page from a case.

    Args:
        case_id: The id of the case.
        page_id: The id of the page to delete.

    Retruns:
        N/A
    """
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/{case_id}/page/{page_id}"
    )
update_page(case_id, page_id, page)

Update a page of a case.

Parameters:

Name Type Description Default
case_id str

The id of the case.

required
page_id str

The id of the page to update.

required
page InputUpdateCasePage

The fields of the page to update.

required
Retruns

N/A

Source code in thehive4py/endpoints/case.py
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
def update_page(
    self, case_id: str, page_id: str, page: InputUpdateCasePage
) -> None:
    """Update a page of a case.

    Args:
        case_id: The id of the case.
        page_id: The id of the page to update.
        page: The fields of the page to update.

    Retruns:
        N/A
    """
    return self._session.make_request(
        "PATCH", path=f"/api/v1/case/{case_id}/page/{page_id}", json=page
    )
find_pages(case_id, filters=None, sortby=None, paginate=None)

Find pages related to a case.

Parameters:

Name Type Description Default
case_id str

The id of the case.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputProcedure]

The list of case pages matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
def find_pages(
    self,
    case_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProcedure]:
    """Find pages related to a case.

    Args:
        case_id: The id of the case.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of case pages matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getCase", "idOrName": case_id},
        {"_name": "pages"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-pages"},
        json={"query": query},
    )
find_attachments(case_id, filters=None, sortby=None, paginate=None)

Find attachments related to a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputAttachment]

The list of case attachments matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
def find_attachments(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputAttachment]:
    """Find attachments related to a case.

    Args:
        case_id: The id of the case.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of case attachments matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getCase", "idOrName": case_id},
        {"_name": "attachments"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-attachments"},
        json={"query": query},
    )
find_comments(case_id, filters=None, sortby=None, paginate=None)

Find comments related to a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
filters Optional[FilterExpr]

The filter expressions to apply in the query.

None
sortby Optional[SortExpr]

The sort expressions to apply in the query.

None
paginate Optional[Paginate]

The pagination experssion to apply in the query.

None

Returns:

Type Description
List[OutputComment]

The list of case comments matched by the query or an empty list.

Source code in thehive4py/endpoints/case.py
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
def find_comments(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputComment]:
    """Find comments related to a case.

    Args:
        case_id: The id of the case.
        filters: The filter expressions to apply in the query.
        sortby: The sort expressions to apply in the query.
        paginate: The pagination experssion to apply in the query.

    Returns:
        The list of case comments matched by the query or an empty list.
    """
    query: QueryExpr = [
        {"_name": "getCase", "idOrName": case_id},
        {"_name": "comments"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-comments"},
        json={"query": query},
    )
close(case_id, status, summary, impact_status='NotApplicable')

Close a case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
status CaseStatusValue

The status to close the case with.

required
summary str

The closure summary of the case.

required
impact_status ImpactStatusValue

The impact status of the case.

'NotApplicable'

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
def close(
    self,
    case_id: CaseId,
    status: CaseStatusValue,
    summary: str,
    impact_status: ImpactStatusValue = "NotApplicable",
) -> None:
    """Close a case.

    Args:
        case_id: The id of the case.
        status: The status to close the case with.
        summary: The closure summary of the case.
        impact_status: The impact status of the case.

    Returns:
        N/A
    """
    case: InputUpdateCase = {
        "status": status,
        "impactStatus": impact_status,
        "summary": summary,
    }
    return self.update(
        case_id,
        case,
    )
open(case_id, status=CaseStatus.InProgress)

Open a closed case.

Parameters:

Name Type Description Default
case_id CaseId

The id of the case.

required
status CaseStatusValue

The status to re-open the case with.

InProgress

Returns:

Type Description
None

N/A

Source code in thehive4py/endpoints/case.py
736
737
738
739
740
741
742
743
744
745
746
747
748
749
def open(
    self, case_id: CaseId, status: CaseStatusValue = CaseStatus.InProgress
) -> None:
    """Open a closed case.

    Args:
        case_id: The id of the case.
        status: The status to re-open the case with.

    Returns:
        N/A
    """
    case: InputUpdateCase = {"status": status}
    return self.update(case_id, case)

case_template

CaseTemplateEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/case_template.py
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputCaseTemplate]:
    query: QueryExpr = [
        {"_name": "listCaseTemplate"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        json={"query": query},
        params={"name": "caseTemplate"},
    )
get(case_template_id)
Source code in thehive4py/endpoints/case_template.py
29
30
31
32
def get(self, case_template_id: str) -> OutputCaseTemplate:
    return self._session.make_request(
        "GET", path=f"/api/v1/caseTemplate/{case_template_id}"
    )
create(case_template)
Source code in thehive4py/endpoints/case_template.py
34
35
36
37
def create(self, case_template: InputCaseTemplate) -> OutputCaseTemplate:
    return self._session.make_request(
        "POST", path="/api/v1/caseTemplate", json=case_template
    )
delete(case_template_id)
Source code in thehive4py/endpoints/case_template.py
39
40
41
42
def delete(self, case_template_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/caseTemplate/{case_template_id}"
    )
update(case_template_id, fields)
Source code in thehive4py/endpoints/case_template.py
44
45
46
47
def update(self, case_template_id: str, fields: InputCaseTemplate) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/caseTemplate/{case_template_id}", json=fields
    )

comment

CommentEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create_in_alert(alert_id, comment)
Source code in thehive4py/endpoints/comment.py
 7
 8
 9
10
def create_in_alert(self, alert_id: str, comment: InputComment) -> OutputComment:
    return self._session.make_request(
        "POST", path=f"/api/v1/alert/{alert_id}/comment", json=comment
    )
create_in_case(case_id, comment)
Source code in thehive4py/endpoints/comment.py
12
13
14
15
def create_in_case(self, case_id: str, comment: InputComment) -> OutputComment:
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/comment", json=comment
    )
get(comment_id)
Source code in thehive4py/endpoints/comment.py
17
18
19
20
21
22
23
24
25
26
27
def get(self, comment_id: str) -> OutputComment:
    # TODO: temp implementation until a dedicated get endpoint
    comments = self._session.make_request(
        "POST",
        path="/api/v1/query",
        json={"query": [{"_name": "getComment", "idOrName": comment_id}]},
    )
    try:
        return comments[0]
    except IndexError:
        raise TheHiveError("404 - Comment not found")
delete(comment_id)
Source code in thehive4py/endpoints/comment.py
29
30
31
32
def delete(self, comment_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/comment/{comment_id}"
    )
update(comment_id, fields)
Source code in thehive4py/endpoints/comment.py
34
35
36
37
def update(self, comment_id: str, fields: InputUpdateComment) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/comment/{comment_id}", json=fields
    )

cortex

CortexEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create_analyzer_job(job)
Source code in thehive4py/endpoints/cortex.py
14
15
16
17
def create_analyzer_job(self, job: InputAnalyzerJob) -> OutputAnalyzerJob:
    return self._session.make_request(
        "POST", path="/api/connector/cortex/job", json=job
    )
create_responder_action(action)
Source code in thehive4py/endpoints/cortex.py
19
20
21
22
23
24
def create_responder_action(
    self, action: InputResponderAction
) -> OutputResponderAction:
    return self._session.make_request(
        "POST", path="/api/connector/cortex/action", json=action
    )
list_analyzers(range=None)
Source code in thehive4py/endpoints/cortex.py
26
27
28
29
30
def list_analyzers(self, range: Optional[str] = None) -> List[OutputAnalyzer]:
    params = {"range": range}
    return self._session.make_request(
        "GET", path="/api/connector/cortex/analyzer", params=params
    )
list_analyzers_by_type(data_type)
Source code in thehive4py/endpoints/cortex.py
32
33
34
35
def list_analyzers_by_type(self, data_type: str) -> List[OutputAnalyzer]:
    return self._session.make_request(
        "GET", path=f"/api/connector/cortex/analyzer/type/{data_type}"
    )
get_analyzer(analyzer_id)
Source code in thehive4py/endpoints/cortex.py
37
38
39
40
def get_analyzer(self, analyzer_id: str) -> OutputAnalyzer:
    return self._session.make_request(
        "GET", path=f"/api/connector/cortex/analyzer/{analyzer_id}"
    )
get_analyzer_job(job_id)
Source code in thehive4py/endpoints/cortex.py
42
43
44
45
def get_analyzer_job(self, job_id: str) -> OutputAnalyzerJob:
    return self._session.make_request(
        "GET", path=f"/api/connector/cortex/job/{job_id}"
    )
list_responders(entity_type, entity_id)
Source code in thehive4py/endpoints/cortex.py
47
48
49
50
51
52
def list_responders(
    self, entity_type: str, entity_id: str
) -> List[OutputResponder]:
    return self._session.make_request(
        "GET", f"/api/connector/cortex/responder/{entity_type}/{entity_id}"
    )

custom_field

CustomFieldEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(custom_field)
Source code in thehive4py/endpoints/custom_field.py
12
13
14
15
def create(self, custom_field: InputCustomField) -> OutputCustomField:
    return self._session.make_request(
        "POST", path="/api/v1/customField", json=custom_field
    )
list()
Source code in thehive4py/endpoints/custom_field.py
17
18
def list(self) -> List[OutputCustomField]:
    return self._session.make_request("GET", path="/api/v1/customField")
delete(custom_field_id)
Source code in thehive4py/endpoints/custom_field.py
20
21
22
23
def delete(self, custom_field_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/customField/{custom_field_id}"
    )
update(custom_field_id, fields)
Source code in thehive4py/endpoints/custom_field.py
25
26
27
28
def update(self, custom_field_id: str, fields: InputUpdateCustomField) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/customField/{custom_field_id}", json=fields
    )

observable

ObservableEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create_in_alert(alert_id, observable, observable_path=None)
Source code in thehive4py/endpoints/observable.py
18
19
20
21
22
23
24
25
26
27
28
29
def create_in_alert(
    self,
    alert_id: str,
    observable: InputObservable,
    observable_path: Optional[str] = None,
) -> List[OutputObservable]:
    kwargs = self._build_observable_kwargs(
        observable=observable, observable_path=observable_path
    )
    return self._session.make_request(
        "POST", path=f"/api/v1/alert/{alert_id}/observable", **kwargs
    )
create_in_case(case_id, observable, observable_path=None)
Source code in thehive4py/endpoints/observable.py
31
32
33
34
35
36
37
38
39
40
41
42
def create_in_case(
    self,
    case_id: str,
    observable: InputObservable,
    observable_path: Optional[str] = None,
) -> List[OutputObservable]:
    kwargs = self._build_observable_kwargs(
        observable=observable, observable_path=observable_path
    )
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/observable", **kwargs
    )
get(observable_id)
Source code in thehive4py/endpoints/observable.py
44
45
46
47
def get(self, observable_id: str) -> OutputObservable:
    return self._session.make_request(
        "GET", path=f"/api/v1/observable/{observable_id}"
    )
delete(observable_id)
Source code in thehive4py/endpoints/observable.py
49
50
51
52
def delete(self, observable_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/observable/{observable_id}"
    )
update(observable_id, fields)
Source code in thehive4py/endpoints/observable.py
54
55
56
57
def update(self, observable_id: str, fields: InputUpdateObservable) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/observable/{observable_id}", json=fields
    )
bulk_update(fields)
Source code in thehive4py/endpoints/observable.py
59
60
61
62
def bulk_update(self, fields: InputBulkUpdateObservable) -> None:
    return self._session.make_request(
        "PATCH", path="/api/v1/observable/_bulk", json=fields
    )
share(observable_id, organisations)
Source code in thehive4py/endpoints/observable.py
64
65
66
67
68
69
def share(self, observable_id: str, organisations: List[str]) -> None:
    return self._session.make_request(
        "POST",
        path=f"/api/v1/observable/{observable_id}/shares",
        json={"organisations": organisations},
    )
unshare(observable_id, organisations)
Source code in thehive4py/endpoints/observable.py
71
72
73
74
75
76
def unshare(self, observable_id: str, organisations: List[str]) -> None:
    return self._session.make_request(
        "DELETE",
        path=f"/api/v1/observable/{observable_id}/shares",
        json={"organisations": organisations},
    )
list_shares(observable_id)
Source code in thehive4py/endpoints/observable.py
78
79
80
81
def list_shares(self, observable_id: str) -> List[OutputShare]:
    return self._session.make_request(
        "GET", path=f"/api/v1/case/{observable_id}/shares"
    )
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/observable.py
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputObservable]:
    query: QueryExpr = [
        {"_name": "listObservable"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "observables"},
        json={"query": query},
    )
count(filters=None)
Source code in thehive4py/endpoints/observable.py
101
102
103
104
105
106
107
108
109
110
111
112
113
def count(self, filters: Optional[FilterExpr] = None) -> int:
    query: QueryExpr = [
        {"_name": "listObservable"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "observable.count"},
        json={"query": query},
    )
download_attachment(observable_id, attachment_id, observable_path, as_zip=False)
Source code in thehive4py/endpoints/observable.py
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def download_attachment(
    self,
    observable_id: str,
    attachment_id: str,
    observable_path: str,
    as_zip=False,
) -> None:
    return self._session.make_request(
        "GET",
        path=(
            f"/api/v1/observable/{observable_id}"
            f"/attachment/{attachment_id}/download"
        ),
        params={"asZip": as_zip},
        download_path=observable_path,
    )

observable_type

ObservableTypeEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(observable_type)
Source code in thehive4py/endpoints/observable_type.py
15
16
17
18
def create(self, observable_type: InputObservableType) -> OutputObservableType:
    return self._session.make_request(
        "POST", path="/api/v1/observable/type", json=observable_type
    )
get(observable_type_id)
Source code in thehive4py/endpoints/observable_type.py
20
21
22
23
def get(self, observable_type_id: str) -> OutputObservableType:
    return self._session.make_request(
        "GET", path=f"/api/v1/observable/type/{observable_type_id}"
    )
delete(observable_type_id)
Source code in thehive4py/endpoints/observable_type.py
25
26
27
28
def delete(self, observable_type_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/observable/type/{observable_type_id}"
    )
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/observable_type.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputObservableType]:
    query: QueryExpr = [
        {"_name": "listObservableType"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "observableTypes"},
        json={"query": query},
    )

organisation

OrganisationEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(organisation)
Source code in thehive4py/endpoints/organisation.py
19
20
21
22
def create(self, organisation: InputOrganisation) -> OutputOrganisation:
    return self._session.make_request(
        "POST", path="/api/v1/organisation", json=organisation
    )
get(org_id)
Source code in thehive4py/endpoints/organisation.py
24
25
def get(self, org_id: str) -> OutputOrganisation:
    return self._session.make_request("GET", path=f"/api/v1/organisation/{org_id}")
update(org_id, fields)
Source code in thehive4py/endpoints/organisation.py
27
28
29
30
def update(self, org_id: str, fields: InputUpdateOrganisation) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/organisation/{org_id}", json=fields
    )
delete(org_id)
Source code in thehive4py/endpoints/organisation.py
32
33
34
35
def delete(self, org_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/organisation/{org_id}"
    )
Source code in thehive4py/endpoints/organisation.py
37
38
39
40
def link(self, org_id: str, other_org_id: str, link: InputOrganisationLink) -> None:
    return self._session.make_request(
        "PUT", path=f"/api/v1/organisation/{org_id}/link/{other_org_id}", json=link
    )
Source code in thehive4py/endpoints/organisation.py
42
43
44
45
def unlink(self, org_id: str, other_org_id: str) -> None:
    return self._session.make_request(
        "GET", path=f"/api/v1/organisation/{org_id}/link/{other_org_id}"
    )
Source code in thehive4py/endpoints/organisation.py
47
48
49
50
def list_links(self, org_id: str) -> List[OutputOrganisation]:
    return self._session.make_request(
        "GET", path=f"/api/v1/organisation/{org_id}/links"
    )
Source code in thehive4py/endpoints/organisation.py
52
53
54
55
def bulk_link(self, org_id: str, links: List[InputBulkOrganisationLink]) -> None:
    return self._session.make_request(
        "PUT", path=f"/api/v1/organisation/{org_id}/links", json={"links": links}
    )
list_sharing_profiles()
Source code in thehive4py/endpoints/organisation.py
57
58
def list_sharing_profiles(self) -> List[OutputSharingProfile]:
    return self._session.make_request("GET", path="/api/v1/sharingProfile")
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/organisation.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputOrganisation]:
    query: QueryExpr = [
        {"_name": "listOrganisation"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "organisations"},
        json={"query": query},
    )
count(filters=None)
Source code in thehive4py/endpoints/organisation.py
78
79
80
81
82
83
84
85
86
87
88
89
90
def count(self, filters: Optional[FilterExpr] = None) -> int:
    query: QueryExpr = [
        {"_name": "listOrganisation"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "organisations.count"},
        json={"query": query},
    )

procedure

ProcedureEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create_in_alert(alert_id, procedure)
Source code in thehive4py/endpoints/procedure.py
17
18
19
20
21
22
def create_in_alert(
    self, alert_id: str, procedure: InputProcedure
) -> OutputProcedure:
    return self._session.make_request(
        "POST", path=f"/api/v1/alert/{alert_id}/procedure", json=procedure
    )
create_in_case(case_id, procedure)
Source code in thehive4py/endpoints/procedure.py
24
25
26
27
28
29
def create_in_case(
    self, case_id: str, procedure: InputProcedure
) -> OutputProcedure:
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/procedure", json=procedure
    )
get(procedure_id)
Source code in thehive4py/endpoints/procedure.py
31
32
33
34
35
36
37
38
39
40
41
def get(self, procedure_id: str) -> OutputProcedure:
    # TODO: temp implementation until a dedicated get endpoint
    procedures = self._session.make_request(
        "POST",
        path="/api/v1/query",
        json={"query": [{"_name": "getProcedure", "idOrName": procedure_id}]},
    )
    try:
        return procedures[0]
    except IndexError:
        raise TheHiveError("404 - Procedure not found")
delete(procedure_id)
Source code in thehive4py/endpoints/procedure.py
43
44
45
46
def delete(self, procedure_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/procedure/{procedure_id}"
    )
update(procedure_id, fields)
Source code in thehive4py/endpoints/procedure.py
48
49
50
51
def update(self, procedure_id: str, fields: InputUpdateProcedure) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/procedure/{procedure_id}", json=fields
    )
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/procedure.py
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProcedure]:
    query: QueryExpr = [
        {"_name": "listProcedure"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "procedures"},
        json={"query": query},
    )

profile

ProfileEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(profile)
Source code in thehive4py/endpoints/profile.py
12
13
def create(self, profile: InputProfile) -> OutputProfile:
    return self._session.make_request("POST", path="/api/v1/profile", json=profile)
get(profile_id)
Source code in thehive4py/endpoints/profile.py
15
16
def get(self, profile_id: str) -> OutputProfile:
    return self._session.make_request("GET", path=f"/api/v1/profile/{profile_id}")
delete(profile_id)
Source code in thehive4py/endpoints/profile.py
18
19
def delete(self, profile_id: str) -> None:
    return self._session.make_request("DELETE", f"/api/v1/profile/{profile_id}")
update(profile_id, fields)
Source code in thehive4py/endpoints/profile.py
21
22
23
24
def update(self, profile_id: str, fields: InputUpdateProfile) -> None:
    return self._session.make_request(
        "PATCH", f"/api/v1/profile/{profile_id}", json=fields
    )
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/profile.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProfile]:
    query: QueryExpr = [
        {"_name": "listProfile"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "profiles"},
        json={"query": query},
    )
count(filters=None)
Source code in thehive4py/endpoints/profile.py
44
45
46
47
48
49
50
51
52
53
54
55
56
def count(self, filters: Optional[FilterExpr] = None) -> int:
    query: QueryExpr = [
        {"_name": "listProfile"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "profile.count"},
        json={"query": query},
    )

query

QueryEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
run(query, exclude_fields=[])
Source code in thehive4py/endpoints/query.py
 7
 8
 9
10
11
12
13
14
def run(
    self, query: typing.List[dict], exclude_fields: typing.List[str] = []
) -> typing.Any:
    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        json={"query": query, "excludeFields": exclude_fields},
    )

task

TaskEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(case_id, task)
Source code in thehive4py/endpoints/task.py
18
19
20
21
def create(self, case_id: str, task: InputTask) -> OutputTask:
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/task", json=task
    )
get(task_id)
Source code in thehive4py/endpoints/task.py
23
24
def get(self, task_id: str) -> OutputTask:
    return self._session.make_request("GET", path=f"/api/v1/task/{task_id}")
delete(task_id)
Source code in thehive4py/endpoints/task.py
26
27
def delete(self, task_id: str) -> None:
    return self._session.make_request("DELETE", path=f"/api/v1/task/{task_id}")
update(task_id, fields)
Source code in thehive4py/endpoints/task.py
29
30
31
32
def update(self, task_id: str, fields: InputUpdateTask) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/task/{task_id}", json=fields
    )
bulk_update(fields)
Source code in thehive4py/endpoints/task.py
34
35
36
37
def bulk_update(self, fields: InputBulkUpdateTask) -> None:
    return self._session.make_request(
        "PATCH", path="/api/v1/task/_bulk", json=fields
    )
get_required_actions(task_id)
Source code in thehive4py/endpoints/task.py
39
40
41
42
def get_required_actions(self, task_id: str) -> dict:
    return self._session.make_request(
        "GET", path=f"/api/v1/task/{task_id}/actionRequired"
    )
set_as_required(task_id, org_id)
Source code in thehive4py/endpoints/task.py
44
45
46
47
def set_as_required(self, task_id: str, org_id: str) -> None:
    return self._session.make_request(
        "PUT", f"/api/v1/task/{task_id}/actionRequired/{org_id}"
    )
set_as_done(task_id, org_id)
Source code in thehive4py/endpoints/task.py
49
50
51
52
def set_as_done(self, task_id: str, org_id: str) -> None:
    return self._session.make_request(
        "PUT", f"/api/v1/task/{task_id}/actionDone/{org_id}"
    )
share()
Source code in thehive4py/endpoints/task.py
54
55
def share(self):
    raise NotImplementedError()
list_shares()
Source code in thehive4py/endpoints/task.py
57
58
def list_shares(self):
    raise NotImplementedError()
unshare()
Source code in thehive4py/endpoints/task.py
60
61
def unshare(self):
    raise NotImplementedError()
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/task.py
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputTask]:
    query: QueryExpr = [
        {"_name": "listTask"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "tasks"},
        json={"query": query},
    )
count(filters=None)
Source code in thehive4py/endpoints/task.py
81
82
83
84
85
86
87
88
89
90
91
92
93
def count(self, filters: Optional[FilterExpr] = None) -> int:
    query: QueryExpr = [
        {"_name": "listTask"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "task.count"},
        json={"query": query},
    )
create_log(task_id, task_log)
Source code in thehive4py/endpoints/task.py
95
96
97
98
def create_log(self, task_id: str, task_log: InputTaskLog) -> OutputTaskLog:
    return self._session.make_request(
        "POST", path=f"/api/v1/task/{task_id}/log", json=task_log
    )
find_logs(task_id, filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/task.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def find_logs(
    self,
    task_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputTaskLog]:
    query: QueryExpr = [
        {"_name": "getTask", "idOrName": task_id},
        {"_name": "logs"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "case-task-logs"},
        json={"query": query},
    )

task_log

TaskLogEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(task_id, task_log)
Source code in thehive4py/endpoints/task_log.py
 9
10
11
12
def create(self, task_id: str, task_log: InputTaskLog) -> OutputTaskLog:
    return self._session.make_request(
        "POST", path=f"/api/v1/task/{task_id}/log", json=task_log
    )
get(task_log_id)
Source code in thehive4py/endpoints/task_log.py
14
15
16
17
18
19
20
21
22
23
24
25
def get(self, task_log_id: str) -> OutputTaskLog:
    # TODO: temp implementation until a dedicated get endpoint
    logs = self._session.make_request(
        "POST",
        path="/api/v1/query",
        json={"query": [{"_name": "getLog", "idOrName": task_log_id}]},
    )

    try:
        return logs[0]
    except IndexError:
        raise TheHiveError("404 - Task Log Not Found")
delete(task_log_id)
Source code in thehive4py/endpoints/task_log.py
27
28
def delete(self, task_log_id: str) -> None:
    return self._session.make_request("DELETE", path=f"/api/v1/log/{task_log_id}")
update(task_log_id, fields)
Source code in thehive4py/endpoints/task_log.py
30
31
32
33
def update(self, task_log_id: str, fields: InputUpdateTaskLog) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/log/{task_log_id}", json=fields
    )
add_attachments(task_log_id, attachment_paths)
Source code in thehive4py/endpoints/task_log.py
35
36
37
38
39
40
41
42
def add_attachments(self, task_log_id: str, attachment_paths: List[str]) -> None:
    files = [
        ("attachments", self._fileinfo_from_filepath(attachment_path))
        for attachment_path in attachment_paths
    ]
    return self._session.make_request(
        "POST", f"/api/v1/log/{task_log_id}/attachments", files=files
    )
delete_attachment(task_log_id, attachment_id)
Source code in thehive4py/endpoints/task_log.py
44
45
46
47
def delete_attachment(self, task_log_id: str, attachment_id: str) -> None:
    return self._session.make_request(
        "DELETE", f"/api/v1/log/{task_log_id}/attachments/{attachment_id}"
    )

timeline

TimelineEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
get(case_id)
Source code in thehive4py/endpoints/timeline.py
11
12
def get(self, case_id: str) -> OutputTimeline:
    return self._session.make_request("GET", f"/api/v1/case/{case_id}/timeline")
create_event(case_id, event)
Source code in thehive4py/endpoints/timeline.py
14
15
16
17
def create_event(self, case_id: str, event: InputCustomEvent) -> OutputCustomEvent:
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/customEvent", json=event
    )
delete_event(event_id)
Source code in thehive4py/endpoints/timeline.py
19
20
21
22
def delete_event(self, event_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/customEvent/{event_id}"
    )
update_event(event_id, fields)
Source code in thehive4py/endpoints/timeline.py
24
25
26
27
def update_event(self, event_id: str, fields: InputUpdateCustomEvent) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/customEvent/{event_id}", json=fields
    )

user

UserEndpoint(session)

Bases: EndpointBase

Source code in thehive4py/endpoints/_base.py
15
16
def __init__(self, session: TheHiveSession):
    self._session = session
create(user)
Source code in thehive4py/endpoints/user.py
18
19
def create(self, user: InputUser) -> OutputUser:
    return self._session.make_request("POST", path="/api/v1/user", json=user)
get(user_id)
Source code in thehive4py/endpoints/user.py
21
22
def get(self, user_id: str) -> OutputUser:
    return self._session.make_request("GET", path=f"/api/v1/user/{user_id}")
get_current()
Source code in thehive4py/endpoints/user.py
24
25
def get_current(self) -> OutputUser:
    return self._session.make_request("GET", path="/api/v1/user/current")
delete(user_id, organisation=None)
Source code in thehive4py/endpoints/user.py
27
28
29
30
31
32
def delete(self, user_id: str, organisation: Optional[str] = None) -> None:
    return self._session.make_request(
        "DELETE",
        path=f"/api/v1/user/{user_id}/force",
        params={"organisation": organisation},
    )
update(user_id, fields)
Source code in thehive4py/endpoints/user.py
34
35
36
37
def update(self, user_id: str, fields: InputUpdateUser) -> None:
    return self._session.make_request(
        "PATCH", path=f"/api/v1/user/{user_id}", json=fields
    )
lock(user_id)
Source code in thehive4py/endpoints/user.py
39
40
def lock(self, user_id: str) -> None:
    return self.update(user_id=user_id, fields={"locked": True})
unlock(user_id)
Source code in thehive4py/endpoints/user.py
42
43
def unlock(self, user_id: str) -> None:
    return self.update(user_id=user_id, fields={"locked": False})
set_organisations(user_id, organisations)
Source code in thehive4py/endpoints/user.py
45
46
47
48
49
50
51
52
def set_organisations(
    self, user_id: str, organisations: List[InputUserOrganisation]
) -> List[OutputUserOrganisation]:
    return self._session.make_request(
        "PUT",
        path=f"/api/v1/user/{user_id}/organisations",
        json={"organisations": organisations},
    )["organisations"]
set_password(user_id, password)
Source code in thehive4py/endpoints/user.py
54
55
56
57
58
59
def set_password(self, user_id: str, password: str) -> None:
    return self._session.make_request(
        "POST",
        path=f"/api/v1/user/{user_id}/password/set",
        json={"password": password},
    )
get_apikey(user_id)
Source code in thehive4py/endpoints/user.py
61
62
def get_apikey(self, user_id: str) -> str:
    return self._session.make_request("GET", path=f"/api/v1/user/{user_id}/key")
remove_apikey(user_id)
Source code in thehive4py/endpoints/user.py
64
65
def remove_apikey(self, user_id: str) -> None:
    return self._session.make_request("DELETE", path=f"/api/v1/user/{user_id}/key")
renew_apikey(user_id)
Source code in thehive4py/endpoints/user.py
67
68
69
70
def renew_apikey(self, user_id: str) -> str:
    return self._session.make_request(
        "POST", path=f"/api/v1/user/{user_id}/key/renew"
    )
get_avatar(user_id)
Source code in thehive4py/endpoints/user.py
72
73
74
def get_avatar(self, user_id: str):
    # TODO: implement the avatar download
    raise NotImplementedError()
find(filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/user.py
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputUser]:
    query: QueryExpr = [
        {"_name": "listUser"},
        *self._build_subquery(filters=filters, sortby=sortby, paginate=paginate),
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "users"},
        json={"query": query},
    )
count(filters=None)
Source code in thehive4py/endpoints/user.py
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def count(self, filters: Optional[FilterExpr] = None) -> int:
    query: QueryExpr = [
        {"_name": "listUser"},
        *self._build_subquery(filters=filters),
        {"_name": "count"},
    ]

    return self._session.make_request(
        "POST",
        path="/api/v1/query",
        params={"name": "cases.count"},
        json={"query": query},
    )

types

alert

InputAlertRequired

Bases: TypedDict

type instance-attribute
source instance-attribute
sourceRef instance-attribute
title instance-attribute
description instance-attribute
InputAlert

Bases: InputAlertRequired

date instance-attribute
severity instance-attribute
tags instance-attribute
flag instance-attribute
tlp instance-attribute
pap instance-attribute
customFields instance-attribute
summary instance-attribute
status instance-attribute
assignee instance-attribute
caseTemplate instance-attribute
observables instance-attribute
procedures instance-attribute
OutputAlertRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
type instance-attribute
source instance-attribute
sourceRef instance-attribute
title instance-attribute
description instance-attribute
severity instance-attribute
severityLabel instance-attribute
date instance-attribute
tlp instance-attribute
tlpLabel instance-attribute
pap instance-attribute
papLabel instance-attribute
follow instance-attribute
observableCount instance-attribute
status instance-attribute
stage instance-attribute
extraData instance-attribute
newDate instance-attribute
timeToDetect instance-attribute
OutputAlert

Bases: OutputAlertRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
tags instance-attribute
customFields instance-attribute
caseTemplate instance-attribute
caseId instance-attribute
assignee instance-attribute
summary instance-attribute
inProgressDate instance-attribute
closedDate instance-attribute
importedDate instance-attribute
timeToTriage instance-attribute
timeToQualify instance-attribute
timeToAcknowledge instance-attribute
InputUpdateAlert

Bases: TypedDict

type instance-attribute
source instance-attribute
sourceRef instance-attribute
title instance-attribute
description instance-attribute
severity instance-attribute
date instance-attribute
lastSyncDate instance-attribute
tags instance-attribute
tlp instance-attribute
pap instance-attribute
follow instance-attribute
customFields instance-attribute
status instance-attribute
summary instance-attribute
assignee instance-attribute
addTags instance-attribute
removeTags instance-attribute
InputBulkUpdateAlert

Bases: InputUpdateAlert

ids instance-attribute
InputPromoteAlert

Bases: TypedDict

title instance-attribute
description instance-attribute
severity instance-attribute
startDate instance-attribute
endDate instance-attribute
tags instance-attribute
flag instance-attribute
tlp instance-attribute
pap instance-attribute
status instance-attribute
summary instance-attribute
assignee instance-attribute
customFields instance-attribute
caseTemplate instance-attribute
tasks instance-attribute
pages instance-attribute
sharingParameters instance-attribute
taskRule instance-attribute
observableRule instance-attribute

attachment

OutputAttachmentRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
name instance-attribute
size instance-attribute
contentType instance-attribute
id instance-attribute
OutputAttachment

Bases: OutputAttachmentRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
hashes instance-attribute

case

CaseStatusValue = Literal['New', 'InProgress', 'Indeterminate', 'FalsePositive', 'TruePositive', 'Other', 'Duplicated'] module-attribute
ImpactStatusValue = Literal['NotApplicable', 'WithImpact', 'NoImpact'] module-attribute
CaseStatus
New = 'New' class-attribute instance-attribute
InProgress = 'InProgress' class-attribute instance-attribute
Indeterminate = 'Indeterminate' class-attribute instance-attribute
FalsePositive = 'FalsePositive' class-attribute instance-attribute
TruePositive = 'TruePositive' class-attribute instance-attribute
Other = 'Other' class-attribute instance-attribute
Duplicated = 'Duplicated' class-attribute instance-attribute
ImpactStatus
NotApplicable = 'NotApplicable' class-attribute instance-attribute
WithImpact = 'WithImpact' class-attribute instance-attribute
NoImpact = 'NoImpact' class-attribute instance-attribute
InputCaseRequired

Bases: TypedDict

title instance-attribute
description instance-attribute
InputCase

Bases: InputCaseRequired

severity instance-attribute
startDate instance-attribute
endDate instance-attribute
tags instance-attribute
flag instance-attribute
tlp instance-attribute
pap instance-attribute
status instance-attribute
summary instance-attribute
assignee instance-attribute
customFields instance-attribute
caseTemplate instance-attribute
tasks instance-attribute
pages instance-attribute
sharingParameters instance-attribute
taskRule instance-attribute
observableRule instance-attribute
OutputCaseRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
number instance-attribute
title instance-attribute
description instance-attribute
severity instance-attribute
severityLabel instance-attribute
startDate instance-attribute
flag instance-attribute
tlp instance-attribute
tlpLabel instance-attribute
pap instance-attribute
papLabel instance-attribute
status instance-attribute
stage instance-attribute
extraData instance-attribute
newDate instance-attribute
timeToDetect instance-attribute
OutputCase

Bases: OutputCaseRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
endDate instance-attribute
tags instance-attribute
summary instance-attribute
impactStatus instance-attribute
assignee instance-attribute
customFields instance-attribute
userPermissions instance-attribute
inProgressDate instance-attribute
closedDate instance-attribute
alertDate instance-attribute
alertNewDate instance-attribute
alertInProgressDate instance-attribute
alertImportedDate instance-attribute
timeToTriage instance-attribute
timeToQualify instance-attribute
timeToAcknowledge instance-attribute
timeToResolve instance-attribute
handlingDuration instance-attribute
InputUpdateCase

Bases: TypedDict

title instance-attribute
description instance-attribute
severity instance-attribute
startDate instance-attribute
endDate instance-attribute
tags instance-attribute
flag instance-attribute
tlp instance-attribute
pap instance-attribute
status instance-attribute
summary instance-attribute
assignee instance-attribute
impactStatus instance-attribute
customFields instance-attribute
taskRule instance-attribute
observableRule instance-attribute
addTags instance-attribute
removeTags instance-attribute
InputBulkUpdateCase

Bases: InputUpdateCase

ids instance-attribute
InputImportCaseRequired

Bases: TypedDict

password instance-attribute
InputImportCase

Bases: InputImportCaseRequired

sharingParameters instance-attribute
taskRule instance-attribute
observableRule instance-attribute
InputApplyCaseTemplateRequired

Bases: TypedDict

ids instance-attribute
caseTemplate instance-attribute
InputApplyCaseTemplate

Bases: InputApplyCaseTemplateRequired

updateTitlePrefix instance-attribute
updateDescription instance-attribute
updateTags instance-attribute
updateSeverity instance-attribute
updateFlag instance-attribute
updateTlp instance-attribute
updatePap instance-attribute
updateCustomFields instance-attribute
importTasks instance-attribute
importPages instance-attribute

case_template

SeverityValue = Literal[1, 2, 3, 4] module-attribute
TlpValue = Literal[0, 1, 2, 3, 4] module-attribute
PapValue = Literal[0, 1, 2, 3] module-attribute
InputCaseTemplateRequired

Bases: TypedDict

name instance-attribute
InputCaseTemplate

Bases: InputCaseTemplateRequired

displayName instance-attribute
titlePrefix instance-attribute
description instance-attribute
severity instance-attribute
tags instance-attribute
flag instance-attribute
tlp instance-attribute
pap instance-attribute
summary instance-attribute
tasks instance-attribute
pageTemplateIds instance-attribute
customFields instance-attribute
OutputCaseTemplateRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
name instance-attribute
OutputCaseTemplate

Bases: OutputCaseTemplateRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
displayName instance-attribute
titlePrefix instance-attribute
description instance-attribute
severity instance-attribute
tags instance-attribute
flag instance-attribute
tlp instance-attribute
pap instance-attribute
summary instance-attribute
tasks instance-attribute
pageTemplateIds instance-attribute
customFields instance-attribute

comment

InputComment

Bases: TypedDict

message instance-attribute
OutputCommentRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
createdBy instance-attribute
createdAt instance-attribute
message instance-attribute
isEdited instance-attribute
OutputComment

Bases: OutputCommentRequired

updatedAt instance-attribute
InputUpdateComment

Bases: TypedDict

message instance-attribute

cortex

OutputAnalyzerRequired

Bases: TypedDict

id instance-attribute
name instance-attribute
version instance-attribute
description instance-attribute
OutputAnalyzer

Bases: OutputAnalyzerRequired

dataTypeList instance-attribute
cortexIds instance-attribute
OutputResponderRequired

Bases: TypedDict

id instance-attribute
name instance-attribute
version instance-attribute
description instance-attribute
OutputResponder

Bases: OutputResponderRequired

dataTypeList instance-attribute
cortexIds instance-attribute
OutputAnalyzerJobRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
analyzerId instance-attribute
analyzerName instance-attribute
analyzerDefinition instance-attribute
status instance-attribute
startDate instance-attribute
cortexId instance-attribute
cortexJobId instance-attribute
id instance-attribute
operations instance-attribute
OutputAnalyzerJob

Bases: OutputAnalyzerJobRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
endDate instance-attribute
report instance-attribute
case_artifact instance-attribute
OutputResponderActionRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
responderId instance-attribute
status instance-attribute
startDate instance-attribute
cortexId instance-attribute
cortexJobId instance-attribute
id instance-attribute
operations instance-attribute
OutputResponderAction

Bases: OutputResponderActionRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
endDate instance-attribute
report instance-attribute
responderName instance-attribute
responderDefinition instance-attribute
InputResponderActionRequired

Bases: TypedDict

objectId instance-attribute
objectType instance-attribute
responderId instance-attribute
InputResponderAction

Bases: InputResponderActionRequired

parameters instance-attribute
tlp instance-attribute
InputAnalyzerJobRequired

Bases: TypedDict

analyzerId instance-attribute
cortexId instance-attribute
artifactId instance-attribute
InputAnalyzerJob

Bases: InputAnalyzerJobRequired

parameters instance-attribute

custom_field

InputCustomFieldValueRequired

Bases: TypedDict

name instance-attribute
InputCustomFieldValue

Bases: InputCustomFieldValueRequired

value instance-attribute
order instance-attribute
OutputCustomFieldValue

Bases: TypedDict

_id instance-attribute
name instance-attribute
description instance-attribute
type instance-attribute
value instance-attribute
order instance-attribute
InputCustomFieldRequired

Bases: TypedDict

name instance-attribute
group instance-attribute
description instance-attribute
type instance-attribute
InputCustomField

Bases: InputCustomFieldRequired

displayName instance-attribute
mandatory instance-attribute
options instance-attribute
OutputCustomFieldRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
name instance-attribute
displayName instance-attribute
group instance-attribute
description instance-attribute
type instance-attribute
mandatory instance-attribute
OutputCustomField

Bases: OutputCustomFieldRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
options instance-attribute
InputUpdateCustomField

Bases: TypedDict

displayName instance-attribute
group instance-attribute
description instance-attribute
type instance-attribute
options instance-attribute
mandatory instance-attribute

observable

InputObservableRequired

Bases: TypedDict

dataType instance-attribute
InputObservable

Bases: InputObservableRequired

data instance-attribute
message instance-attribute
startDate instance-attribute
tlp instance-attribute
pap instance-attribute
tags instance-attribute
ioc instance-attribute
sighted instance-attribute
sightedAt instance-attribute
ignoreSimilarity instance-attribute
isZip instance-attribute
zipPassword instance-attribute
attachment instance-attribute
OutputObservableRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
dataType instance-attribute
startDate instance-attribute
tlp instance-attribute
pap instance-attribute
ioc instance-attribute
sighted instance-attribute
reports instance-attribute
extraData instance-attribute
ignoreSimilarity instance-attribute
OutputObservable

Bases: OutputObservableRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
data instance-attribute
attachment instance-attribute
tags instance-attribute
sightedAt instance-attribute
message instance-attribute
InputUpdateObservable

Bases: TypedDict

dataType instance-attribute
message instance-attribute
tlp instance-attribute
pap instance-attribute
tags instance-attribute
ioc instance-attribute
sighted instance-attribute
sightedAt instance-attribute
ignoreSimilarity instance-attribute
InputBulkUpdateObservable

Bases: InputUpdateObservable

ids instance-attribute

observable_type

InputObservableTypeRequired

Bases: TypedDict

name instance-attribute
InputObservableType

Bases: InputObservableTypeRequired

isAttachment instance-attribute
OutputObservableTypeRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
name instance-attribute
isAttachment instance-attribute
OutputObservableType

Bases: OutputObservableTypeRequired

_updatedBy instance-attribute
_updatedAt instance-attribute

organisation

Bases: TypedDict

linkType instance-attribute
otherLinkType instance-attribute

Bases: TypedDict

toOrganisation instance-attribute
linkType instance-attribute
otherLinkType instance-attribute
OutputSharingProfile

Bases: TypedDict

name instance-attribute
description instance-attribute
autoShare instance-attribute
editable instance-attribute
permissionProfile instance-attribute
taskRule instance-attribute
observableRule instance-attribute
InputOrganisationRequired

Bases: TypedDict

name instance-attribute
description instance-attribute
InputOrganisation

Bases: InputOrganisationRequired

taskRule instance-attribute
observableRule instance-attribute
locked instance-attribute
OutputOrganisationRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
name instance-attribute
description instance-attribute
taskRule instance-attribute
observableRule instance-attribute
locked instance-attribute
extraData instance-attribute
OutputOrganisation

Bases: OutputOrganisationRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
avatar instance-attribute
InputUpdateOrganisation

Bases: TypedDict

name instance-attribute
description instance-attribute
taskRule instance-attribute
observableRule instance-attribute
locked instance-attribute
avatar instance-attribute

page

InputCasePageRequired

Bases: TypedDict

title instance-attribute
content instance-attribute
category instance-attribute
InputCasePage

Bases: InputCasePageRequired

order instance-attribute
OutputCasePageRequired

Bases: TypedDict

_id instance-attribute
id instance-attribute
createdBy instance-attribute
createdAt instance-attribute
title instance-attribute
content instance-attribute
_type instance-attribute
slug instance-attribute
order instance-attribute
category instance-attribute
OutputCasePage

Bases: OutputCasePageRequired

updatedBy instance-attribute
updatedAt instance-attribute
InputUpdateCasePage

Bases: TypedDict

title instance-attribute
content instance-attribute
category instance-attribute
order instance-attribute

procedure

InputProcedureRequired

Bases: TypedDict

occurDate instance-attribute
patternId instance-attribute
InputProcedure

Bases: InputProcedureRequired

tactic instance-attribute
description instance-attribute
OutputProcedureRequired

Bases: TypedDict

_id instance-attribute
_createdAt instance-attribute
_createdBy instance-attribute
occurDate instance-attribute
tactic instance-attribute
tacticLabel instance-attribute
extraData instance-attribute
OutputProcedure

Bases: OutputProcedureRequired

_updatedAt instance-attribute
_updatedBy instance-attribute
description instance-attribute
patternId instance-attribute
patternName instance-attribute
InputUpdateProcedure

Bases: TypedDict

description instance-attribute
occurDate instance-attribute

profile

InputProfileRequired

Bases: TypedDict

name instance-attribute
InputProfile

Bases: InputProfileRequired

permissions instance-attribute
OutputProfileRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
name instance-attribute
editable instance-attribute
isAdmin instance-attribute
OutputProfile

Bases: OutputProfileRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
permissions instance-attribute
InputUpdateProfile

Bases: TypedDict

name instance-attribute
permissions instance-attribute

share

OutputShareRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
caseId instance-attribute
profileName instance-attribute
organisationName instance-attribute
owner instance-attribute
taskRule instance-attribute
observableRule instance-attribute
OutputShare

Bases: OutputShareRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
InputShareRequired

Bases: TypedDict

organisation instance-attribute
InputShare

Bases: InputShareRequired

share instance-attribute
profile instance-attribute
taskRule instance-attribute
observableRule instance-attribute

task

InputTaskRequired

Bases: TypedDict

title instance-attribute
InputTask

Bases: InputTaskRequired

group instance-attribute
description instance-attribute
status instance-attribute
flag instance-attribute
startDate instance-attribute
endDate instance-attribute
order instance-attribute
dueDate instance-attribute
assignee instance-attribute
mandatory instance-attribute
OutputTaskRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
title instance-attribute
group instance-attribute
status instance-attribute
flag instance-attribute
order instance-attribute
mandatory instance-attribute
extraData instance-attribute
OutputTask

Bases: OutputTaskRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
description instance-attribute
startDate instance-attribute
endDate instance-attribute
assignee instance-attribute
dueDate instance-attribute
InputUpdateTask

Bases: TypedDict

title instance-attribute
group instance-attribute
description instance-attribute
status instance-attribute
flag instance-attribute
startDate instance-attribute
endDate instance-attribute
order instance-attribute
dueDate instance-attribute
assignee instance-attribute
mandatory instance-attribute
InputBulkUpdateTask

Bases: InputUpdateTask

ids instance-attribute

task_log

InputTaskLogRequired

Bases: TypedDict

message instance-attribute
InputTaskLog

Bases: InputTaskLogRequired

startDate instance-attribute
includeInTimeline instance-attribute
OutputTaskLogRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
message instance-attribute
date instance-attribute
owner instance-attribute
extraData instance-attribute
OutputTaskLog

Bases: OutputTaskLogRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
attachments instance-attribute
includeInTimeline instance-attribute
InputUpdateTaskLog

Bases: TypedDict

message instance-attribute
includeInTimeline instance-attribute

timeline

OutputTimelineEventRequired

Bases: TypedDict

date instance-attribute
kind instance-attribute
entity instance-attribute
entityId instance-attribute
details instance-attribute
OutputTimelineEvent

Bases: OutputTimelineEventRequired

endDate instance-attribute
OutputTimeline

Bases: TypedDict

events instance-attribute
InputCustomEventRequired

Bases: TypedDict

date instance-attribute
title instance-attribute
InputCustomEvent

Bases: InputCustomEventRequired

endDate instance-attribute
description instance-attribute
OutputCustomEventRequired

Bases: TypedDict

_id instance-attribute
_type instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
date instance-attribute
title instance-attribute
OutputCustomEvent

Bases: OutputCustomEventRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
endDate instance-attribute
description instance-attribute
InputUpdateCustomEvent

Bases: TypedDict

date instance-attribute
endDate instance-attribute
title instance-attribute
description instance-attribute

user

InputUserRequired

Bases: TypedDict

login instance-attribute
name instance-attribute
profile instance-attribute
InputUser

Bases: InputUserRequired

email instance-attribute
password instance-attribute
organisation instance-attribute
type instance-attribute
OutputOrganisationProfile

Bases: TypedDict

organisationId instance-attribute
organisation instance-attribute
profile instance-attribute
OutputUserRequired

Bases: TypedDict

_id instance-attribute
_createdBy instance-attribute
_createdAt instance-attribute
login instance-attribute
name instance-attribute
hasKey instance-attribute
hasPassword instance-attribute
hasMFA instance-attribute
locked instance-attribute
profile instance-attribute
organisation instance-attribute
type instance-attribute
extraData instance-attribute
OutputUser

Bases: OutputUserRequired

_updatedBy instance-attribute
_updatedAt instance-attribute
email instance-attribute
permissions instance-attribute
avatar instance-attribute
organisations instance-attribute
defaultOrganisation instance-attribute
InputUpdateUser

Bases: TypedDict

name instance-attribute
organisation instance-attribute
profile instance-attribute
locked instance-attribute
avatar instance-attribute
email instance-attribute
defaultOrganisation instance-attribute
InputUserOrganisationRequired

Bases: TypedDict

organisation instance-attribute
profile instance-attribute
InputUserOrganisation

Bases: InputUserOrganisationRequired

default instance-attribute
OutputUserOrganisation

Bases: TypedDict

organisation instance-attribute
profile instance-attribute
default instance-attribute

query

QueryExpr = List[Union[_FilterExpr, _SortExpr, Paginate, dict]] module-attribute

filters

FilterExpr = _Union['_FilterBase', dict] module-attribute
_FilterBase

Bases: UserDict

Base class for filters.

__and__(other)
Source code in thehive4py/query/filters.py
12
13
14
15
16
def __and__(self, other: "_FilterBase") -> "_FilterBase":
    if not isinstance(other, _FilterBase):
        self._raise_type_error("&", self, other)
    args = self.get("_and", [self]) + other.get("_and", [other])
    return _FilterBase(_and=args)
__or__(other)
Source code in thehive4py/query/filters.py
18
19
20
21
22
def __or__(self, other: "_FilterBase") -> "_FilterBase":  # type:ignore
    if not isinstance(other, _FilterBase):
        self._raise_type_error("|", self, other)
    args = self.get("_or", [self]) + other.get("_or", [other])
    return _FilterBase(_or=args)
__invert__()
Source code in thehive4py/query/filters.py
24
25
def __invert__(self) -> "_FilterBase":
    return _FilterBase(_not=self)
_raise_type_error(operand, first, second)
Source code in thehive4py/query/filters.py
27
28
29
30
31
def _raise_type_error(self, operand, first, second):
    raise TypeError(
        f"unsupported operand type(s) for {operand}: "
        f"{type(first)} and {type(second)}"
    )
Lt(field, value)

Bases: _FilterBase

Field less than value.

Source code in thehive4py/query/filters.py
37
38
def __init__(self, field: str, value: _Any):
    super().__init__(_lt={"_field": field, "_value": value})
Gt(field, value)

Bases: _FilterBase

Field greater than value.

Source code in thehive4py/query/filters.py
44
45
def __init__(self, field: str, value: _Any):
    super().__init__(_gt={"_field": field, "_value": value})
Lte(field, value)

Bases: _FilterBase

Field less than or equal value.

Source code in thehive4py/query/filters.py
51
52
def __init__(self, field: str, value: _Any):
    super().__init__(_lte={"_field": field, "_value": value})
Gte(field, value)

Bases: _FilterBase

Field less than or equal value.

Source code in thehive4py/query/filters.py
58
59
def __init__(self, field: str, value: _Any):
    super().__init__(_gte={"_field": field, "_value": value})
Ne(field, value)

Bases: _FilterBase

Field not equal value.

Source code in thehive4py/query/filters.py
65
66
def __init__(self, field: str, value: _Any):
    super().__init__(_ne={"_field": field, "_value": value})
Eq(field, value)

Bases: _FilterBase

Field equal value.

Source code in thehive4py/query/filters.py
72
73
def __init__(self, field: str, value: _Any):
    super().__init__(_eq={"_field": field, "_value": value})
StartsWith(field, value)

Bases: _FilterBase

Field starts with value.

Source code in thehive4py/query/filters.py
79
80
def __init__(self, field: str, value: str):
    super().__init__(_startsWith={"_field": field, "_value": value})
EndsWith(field, value)

Bases: _FilterBase

Field ends with value.

Source code in thehive4py/query/filters.py
86
87
def __init__(self, field: str, value: str):
    super().__init__(_endsWith={"_field": field, "_value": value})
Id(id)

Bases: _FilterBase

FIlter by ID.

Source code in thehive4py/query/filters.py
93
94
def __init__(self, id: str):
    super().__init__(_id=id)
Between(field, start, end)

Bases: _FilterBase

Field between inclusive from and exclusive to values.

Source code in thehive4py/query/filters.py
100
101
def __init__(self, field: str, start: int, end: int):
    super().__init__(_between={"_field": field, "_from": start, "_to": end})
In(field, values)

Bases: _FilterBase

Field is one of the values.

Source code in thehive4py/query/filters.py
107
108
def __init__(self, field: _Any, values: list):
    super().__init__(_in={"_field": field, "_values": values})
Contains(field)

Bases: _FilterBase

Object contains the field.

Source code in thehive4py/query/filters.py
114
115
116
117
118
119
120
121
122
def __init__(self, field: str):
    warnings.warn(
        message="The `Contains` filter has been deprecated. "
        "Please use the `Has` filter to prevent breaking "
        "changes in the future.",
        category=DeprecationWarning,
        stacklevel=2,
    )
    super().__init__(_contains=field)
Has(field)

Bases: _FilterBase

Object contains the field.

Source code in thehive4py/query/filters.py
128
129
def __init__(self, field: str):
    super().__init__(_has=field)
Like(field, value)

Bases: _FilterBase

Field contains the value.

Source code in thehive4py/query/filters.py
135
136
def __init__(self, field: str, value: str):
    super().__init__(_like={"_field": field, "_value": value})
Match(field, value)

Bases: _FilterBase

Field contains the value

Source code in thehive4py/query/filters.py
142
143
def __init__(self, field: str, value: str):
    super().__init__(_match={"_field": field, "_value": value})

page

Paginate(start, end, extra_data=[])

Bases: UserDict

Source code in thehive4py/query/page.py
5
6
def __init__(self, start: int, end: int, extra_data=[]):
    super().__init__({"from": start, "to": end, "extraData": extra_data})

sort

SortExpr

Bases: UserDict

Base class for sort expressions.

__and__(other)
Source code in thehive4py/query/sort.py
7
8
def __and__(self, other: "SortExpr") -> "SortExpr":
    return self._concat_expressions("&", self, other)
__or__(other)
Source code in thehive4py/query/sort.py
10
11
def __or__(self, other: "SortExpr") -> "SortExpr":  # type:ignore
    return self._concat_expressions("|", self, other)
_concat_expressions(operand, expr1, expr2)
Source code in thehive4py/query/sort.py
13
14
15
16
17
18
def _concat_expressions(
    self, operand: str, expr1: "SortExpr", expr2: "SortExpr"
) -> "SortExpr":
    if not isinstance(expr1, SortExpr) or not isinstance(expr2, SortExpr):
        self._raise_type_error(operand, expr1, expr2)
    return SortExpr(_fields=[*expr1["_fields"], *expr2["_fields"]])
_raise_type_error(operand, first, second)
Source code in thehive4py/query/sort.py
20
21
22
23
24
def _raise_type_error(self, operand, first, second):
    raise TypeError(
        f"unsupported operand type(s) for {operand}: "
        f"{type(first)} and {type(second)}"
    )
Asc(field)

Bases: SortExpr

Source code in thehive4py/query/sort.py
28
29
def __init__(self, field: str):
    super().__init__(_fields=[{field: "asc"}])
Desc(field)

Bases: SortExpr

Source code in thehive4py/query/sort.py
33
34
def __init__(self, field: str):
    super().__init__(_fields=[{field: "desc"}])

errors

TheHiveError(message, response=None, *args, **kwargs)

Bases: Exception

Base error class of thehive4py.

Parameters:

Name Type Description Default
message str

The exception message.

required
response Optional[Response]

Either None, or a Response object of a failed request.

None
Source code in thehive4py/errors.py
 8
 9
10
11
12
13
14
15
16
17
18
19
def __init__(
    self, message: str, response: Optional[Response] = None, *args, **kwargs
):
    """Base error class of thehive4py.

    Args:
        message: The exception message.
        response: Either `None`, or a `Response` object of a failed request.
    """
    super().__init__(message, *args, **kwargs)
    self.message = message
    self.response = response
message = message instance-attribute
response = response instance-attribute

helpers

now_to_ts()

Return now as TheHive timestamp.

Source code in thehive4py/helpers.py
6
7
8
def now_to_ts() -> int:
    """Return now as TheHive timestamp."""
    return int(time.time() * 1000)

dt_to_ts(datetime)

Convert datetime object to TheHive timestamp.

Source code in thehive4py/helpers.py
11
12
13
def dt_to_ts(datetime: dt.datetime) -> int:
    """Convert datetime object to TheHive timestamp."""
    return int(datetime.timestamp() * 1000)

ts_to_dt(timestamp, tz=None)

Convert TheHive timestamp to datetime object.

Source code in thehive4py/helpers.py
16
17
18
def ts_to_dt(timestamp: int, tz: Optional[dt.timezone] = None) -> dt.datetime:
    """Convert TheHive timestamp to datetime object."""
    return dt.datetime.fromtimestamp(timestamp / 1000.0, tz=tz)