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
24
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
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.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
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: Optional[str] 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

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
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)

endpoints

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)
Source code in thehive4py/endpoints/case.py
34
35
def create(self, case: InputCase) -> OutputCase:
    return self._session.make_request("POST", path="/api/v1/case", json=case)
get(case_id)
Source code in thehive4py/endpoints/case.py
37
38
def get(self, case_id: CaseId) -> OutputCase:
    return self._session.make_request("GET", path=f"/api/v1/case/{case_id}")
delete(case_id)
Source code in thehive4py/endpoints/case.py
40
41
def delete(self, case_id: CaseId) -> None:
    self._session.make_request("DELETE", path=f"/api/v1/case/{case_id}")
update(case_id, fields={}, **kwargs)
Source code in thehive4py/endpoints/case.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def update(
    self, case_id: CaseId, fields: Optional[InputUpdateCase] = {}, **kwargs
) -> None:

    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)
Source code in thehive4py/endpoints/case.py
66
67
68
69
def bulk_update(self, fields: InputBulkUpdateCase) -> None:
    return self._session.make_request(
        "PATCH", path="/api/v1/case/_bulk", json=fields
    )
merge(case_ids)
Source code in thehive4py/endpoints/case.py
71
72
73
74
75
def merge(self, case_ids: Sequence[CaseId]) -> OutputCase:
    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}"
    )
Source code in thehive4py/endpoints/case.py
77
78
79
80
def unlink_alert(self, case_id: str, alert_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/{case_id}/alert/{alert_id}"
    )
merge_similar_observables(case_id)
Source code in thehive4py/endpoints/case.py
82
83
84
85
86
def merge_similar_observables(self, case_id: CaseId) -> dict:
    # 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)
Source code in thehive4py/endpoints/case.py
88
89
def get_linked_cases(self, case_id: CaseId) -> List[OutputCase]:
    return self._session.make_request("GET", path=f"/api/v1/case/{case_id}/links")
delete_custom_field(custom_field_id)
Source code in thehive4py/endpoints/case.py
91
92
93
94
def delete_custom_field(self, custom_field_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/customField/{custom_field_id}"
    )
import_from_file(import_case, import_path)
Source code in thehive4py/endpoints/case.py
 96
 97
 98
 99
100
101
102
103
def import_from_file(self, import_case: InputImportCase, import_path: str) -> dict:
    # 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)
Source code in thehive4py/endpoints/case.py
105
106
107
108
109
110
111
def export_to_file(self, case_id: CaseId, password: str, export_path: str) -> None:
    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)
Source code in thehive4py/endpoints/case.py
113
114
def get_timeline(self, case_id: CaseId) -> OutputTimeline:
    return self._session.make_request("GET", f"/api/v1/case/{case_id}/timeline")
add_attachment(case_id, attachment_paths)
Source code in thehive4py/endpoints/case.py
116
117
118
119
120
121
122
123
124
125
def add_attachment(
    self, case_id: CaseId, attachment_paths: List[str]
) -> List[OutputAttachment]:
    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)
Source code in thehive4py/endpoints/case.py
127
128
129
130
131
132
133
134
def download_attachment(
    self, case_id: CaseId, attachment_id: str, attachment_path: str
) -> None:
    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)
Source code in thehive4py/endpoints/case.py
136
137
138
139
def delete_attachment(self, case_id: CaseId, attachment_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/{case_id}/attachment/{attachment_id}"
    )
list_shares(case_id)
Source code in thehive4py/endpoints/case.py
141
142
def list_shares(self, case_id: CaseId) -> List[OutputShare]:
    return self._session.make_request("GET", path=f"/api/v1/case/{case_id}/shares")
share(case_id, shares)
Source code in thehive4py/endpoints/case.py
144
145
146
147
def share(self, case_id: CaseId, shares: List[InputShare]) -> List[OutputShare]:
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/shares", json={"shares": shares}
    )
unshare(case_id, organisation_ids)
Source code in thehive4py/endpoints/case.py
149
150
151
152
153
154
def unshare(self, case_id: CaseId, organisation_ids: List[str]) -> None:
    return self._session.make_request(
        "DELETE",
        path=f"/api/v1/case/{case_id}/shares",
        json={"organisations": organisation_ids},
    )
set_share(case_id, shares)
Source code in thehive4py/endpoints/case.py
156
157
158
159
def set_share(self, case_id: CaseId, shares: List[InputShare]) -> List[OutputShare]:
    return self._session.make_request(
        "PUT", path=f"/api/v1/case/{case_id}/shares", json={"shares": shares}
    )
remove_share(share_id)
Source code in thehive4py/endpoints/case.py
161
162
163
164
def remove_share(self, share_id: str) -> None:
    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
166
167
168
169
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)
Source code in thehive4py/endpoints/case.py
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def find(
    self,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputCase]:
    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)
Source code in thehive4py/endpoints/case.py
189
190
191
192
193
194
195
196
197
198
199
200
201
def count(self, filters: Optional[FilterExpr] = None) -> int:
    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)
Source code in thehive4py/endpoints/case.py
203
204
205
206
207
208
def create_task(self, case_id: CaseId, task: InputTask) -> OutputTask:
    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)
Source code in thehive4py/endpoints/case.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def find_tasks(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputTask]:
    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)
Source code in thehive4py/endpoints/case.py
230
231
232
233
234
235
236
237
238
239
240
241
def create_observable(
    self,
    case_id: CaseId,
    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
    )
find_observables(case_id, filters=None, sortby=None, paginate=None)
Source code in thehive4py/endpoints/case.py
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
def find_observables(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputObservable]:
    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)
Source code in thehive4py/endpoints/case.py
262
263
264
265
266
267
def create_procedure(
    self, case_id: str, procedure: InputProcedure
) -> OutputProcedure:
    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)
Source code in thehive4py/endpoints/case.py
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def find_procedures(
    self,
    case_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProcedure]:
    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)
Source code in thehive4py/endpoints/case.py
289
290
291
292
def create_page(self, case_id: str, page: InputCasePage) -> OutputCasePage:
    return self._session.make_request(
        "POST", path=f"/api/v1/case/{case_id}/page", json=page
    )
delete_page(case_id, page_id)
Source code in thehive4py/endpoints/case.py
294
295
296
297
def delete_page(self, case_id: str, page_id: str) -> None:
    return self._session.make_request(
        "DELETE", path=f"/api/v1/case/{case_id}/page/{page_id}"
    )
update_page(case_id, page_id, page)
Source code in thehive4py/endpoints/case.py
299
300
301
302
303
304
def update_page(
    self, case_id: str, page_id: str, page: InputUpdateCasePage
) -> None:
    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)
Source code in thehive4py/endpoints/case.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def find_pages(
    self,
    case_id: str,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputProcedure]:
    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)
Source code in thehive4py/endpoints/case.py
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
def find_attachments(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputAttachment]:
    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)
Source code in thehive4py/endpoints/case.py
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def find_comments(
    self,
    case_id: CaseId,
    filters: Optional[FilterExpr] = None,
    sortby: Optional[SortExpr] = None,
    paginate: Optional[Paginate] = None,
) -> List[OutputComment]:
    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')
Source code in thehive4py/endpoints/case.py
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
def close(
    self,
    case_id: CaseId,
    status: CaseStatusValue,
    summary: str,
    impact_status: ImpactStatusValue = "NotApplicable",
) -> None:
    case: InputUpdateCase = {
        "status": status,
        "impactStatus": impact_status,
        "summary": summary,
    }
    return self.update(
        case_id,
        case,
    )
open(case_id, status=CaseStatus.InProgress)
Source code in thehive4py/endpoints/case.py
381
382
383
384
385
def open(
    self, case_id: CaseId, status: CaseStatusValue = CaseStatus.InProgress
) -> None:
    case: InputUpdateCase = {"status": status}
    return self.update(case_id, case)

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(cortex_id, analyzer_id, observable_id)
Source code in thehive4py/endpoints/cortex.py
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def create_analyzer_job(
    self, cortex_id: str, analyzer_id: str, observable_id: str
) -> dict:
    return self._session.make_request(
        "POST",
        path="/api/connector/cortex/job",
        json={
            "analyzerId": analyzer_id,
            "cortexId": cortex_id,
            "artifactId": observable_id,
        },
    )
create_responder_action(object_id, object_type, responder_id)
Source code in thehive4py/endpoints/cortex.py
18
19
20
21
22
23
24
25
26
27
28
29
def create_responder_action(
    self, object_id: str, object_type: str, responder_id: str
) -> dict:
    return self._session.make_request(
        "POST",
        path="/api/connector/cortex/action",
        json={
            "objectId": object_id,
            "objectType": object_type,
            "responderId": responder_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: str instance-attribute
source: str instance-attribute
sourceRef: str instance-attribute
title: str instance-attribute
description: str instance-attribute
InputAlert

Bases: InputAlertRequired

date: int instance-attribute
severity: int instance-attribute
tags: List[str] instance-attribute
flag: bool instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
customFields: List[InputCustomFieldValue] instance-attribute
summary: str instance-attribute
status: str instance-attribute
caseTemplate: str instance-attribute
observables: List[InputObservable] instance-attribute
procedures: List[InputProcedure] instance-attribute
OutputAlertRequired

Bases: TypedDict

type: str instance-attribute
source: str instance-attribute
sourceRef: str instance-attribute
title: str instance-attribute
description: str instance-attribute
severity: int instance-attribute
date: int instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
follow: bool instance-attribute
observableCount: int instance-attribute
status: str instance-attribute
stage: str instance-attribute
extraData: dict instance-attribute
newDate: int instance-attribute
timeToDetect: int instance-attribute
OutputAlert

Bases: OutputAlertRequired

tags: List[str] instance-attribute
customFields: List[OutputCustomFieldValue] instance-attribute
caseTemplate: str instance-attribute
caseId: str instance-attribute
summary: str instance-attribute
inProgressDate: int instance-attribute
closedDate: int instance-attribute
importedDate: int instance-attribute
timeToTriage: int instance-attribute
timeToQualify: int instance-attribute
timeToAcknowledge: int instance-attribute
InputUpdateAlert

Bases: TypedDict

type: str instance-attribute
source: str instance-attribute
sourceRef: str instance-attribute
title: str instance-attribute
description: str instance-attribute
severity: int instance-attribute
date: int instance-attribute
lastSyncDate: int instance-attribute
tags: List[str] instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
follow: bool instance-attribute
customFields: List[InputCustomFieldValue] instance-attribute
status: str instance-attribute
summary: str instance-attribute
InputBulkUpdateAlert

Bases: InputUpdateAlert

ids: List[str] instance-attribute
InputPromoteAlert

Bases: TypedDict

title: str instance-attribute
description: str instance-attribute
severity: int instance-attribute
startDate: int instance-attribute
endDate: int instance-attribute
tags: List[str] instance-attribute
flag: bool instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
status: str instance-attribute
summary: str instance-attribute
assignee: str instance-attribute
customFields: List[InputCustomFieldValue] instance-attribute
caseTemplate: str instance-attribute
tasks: List[InputTask] instance-attribute
sharingParameters: List[InputShare] instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute

attachment

OutputAttachmentRequired

Bases: TypedDict

name: str instance-attribute
size: int instance-attribute
contentType: str instance-attribute
id: str instance-attribute
OutputAttachment

Bases: OutputAttachmentRequired

hashes: List[str] instance-attribute

case

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

Bases: TypedDict

title: str instance-attribute
description: str instance-attribute
InputCase

Bases: InputCaseRequired

severity: int instance-attribute
startDate: int instance-attribute
endDate: int instance-attribute
tags: List[str] instance-attribute
flag: bool instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
status: CaseStatusValue instance-attribute
summary: str instance-attribute
assignee: str instance-attribute
customFields: List[InputCustomFieldValue] instance-attribute
caseTemplate: str instance-attribute
tasks: List[InputTask] instance-attribute
sharingParameters: List[InputShare] instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute
OutputCaseRequired

Bases: TypedDict

number: int instance-attribute
title: str instance-attribute
description: str instance-attribute
severity: int instance-attribute
startDate: int instance-attribute
flag: bool instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
status: CaseStatusValue instance-attribute
stage: str instance-attribute
extraData: dict instance-attribute
newDate: int instance-attribute
timeToDetect: int instance-attribute
OutputCase

Bases: OutputCaseRequired

endDate: int instance-attribute
tags: List[str] instance-attribute
summary: str instance-attribute
impactStatus: ImpactStatusValue instance-attribute
assignee: str instance-attribute
customFields: List[OutputCustomFieldValue] instance-attribute
userPermissions: List[str] instance-attribute
inProgressDate: int instance-attribute
closedDate: int instance-attribute
alertDate: int instance-attribute
alertNewDate: int instance-attribute
alertInProgressDate: int instance-attribute
alertImportedDate: int instance-attribute
timeToTriage: int instance-attribute
timeToQualify: int instance-attribute
timeToAcknowledge: int instance-attribute
timeToResolve: int instance-attribute
handlingDuration: int instance-attribute
InputUpdateCase

Bases: TypedDict

title: str instance-attribute
description: str instance-attribute
severity: int instance-attribute
startDate: int instance-attribute
endDate: int instance-attribute
tags: List[str] instance-attribute
flag: bool instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
status: str instance-attribute
summary: str instance-attribute
assignee: str instance-attribute
impactStatus: str instance-attribute
customFields: List[InputCustomFieldValue] instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute
InputBulkUpdateCase

Bases: InputUpdateCase

ids: List[str] instance-attribute
InputImportCaseRequired

Bases: TypedDict

password: str instance-attribute
InputImportCase

Bases: InputImportCaseRequired

sharingParameters: List[InputShare] instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute

comment

InputComment

Bases: TypedDict

message: str instance-attribute
OutputCommentRequired

Bases: TypedDict

createdBy: str instance-attribute
createdAt: int instance-attribute
message: str instance-attribute
isEdited: bool instance-attribute
OutputComment

Bases: OutputCommentRequired

updatedAt: str instance-attribute
InputUpdateComment

Bases: TypedDict

message: str instance-attribute

custom_field

InputCustomFieldValueRequired

Bases: TypedDict

name: str instance-attribute
InputCustomFieldValue

Bases: InputCustomFieldValueRequired

value: Any instance-attribute
order: int instance-attribute
OutputCustomFieldValue

Bases: TypedDict

name: str instance-attribute
description: str instance-attribute
type: str instance-attribute
value: Any instance-attribute
order: int instance-attribute
InputCustomFieldRequired

Bases: TypedDict

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

Bases: InputCustomFieldRequired

displayName: str instance-attribute
mandatory: bool instance-attribute
options: list instance-attribute
OutputCustomFieldRequired

Bases: TypedDict

name: str instance-attribute
displayName: str instance-attribute
group: str instance-attribute
description: str instance-attribute
type: str instance-attribute
mandatory: bool instance-attribute
OutputCustomField

Bases: OutputCustomFieldRequired

options: list instance-attribute
InputUpdateCustomField

Bases: TypedDict

displayName: str instance-attribute
group: str instance-attribute
description: str instance-attribute
type: str instance-attribute
options: list instance-attribute
mandatory: bool instance-attribute

observable

InputObservableRequired

Bases: TypedDict

dataType: str instance-attribute
InputObservable

Bases: InputObservableRequired

data: str instance-attribute
message: str instance-attribute
startDate: int instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
tags: List[str] instance-attribute
ioc: bool instance-attribute
sighted: bool instance-attribute
sightedAt: int instance-attribute
ignoreSimilarity: bool instance-attribute
isZip: bool instance-attribute
zipPassword: bool instance-attribute
attachment: str instance-attribute
OutputObservableRequired

Bases: TypedDict

dataType: str instance-attribute
startDate: int instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
ioc: bool instance-attribute
sighted: bool instance-attribute
reports: dict instance-attribute
extraData: dict instance-attribute
ignoreSimilarity: bool instance-attribute
OutputObservable

Bases: OutputObservableRequired

data: str instance-attribute
attachment: OutputAttachment instance-attribute
tags: List[str] instance-attribute
sightedAt: int instance-attribute
message: str instance-attribute
InputUpdateObservable

Bases: TypedDict

dataType: str instance-attribute
message: str instance-attribute
tlp: int instance-attribute
pap: int instance-attribute
tags: List[str] instance-attribute
ioc: bool instance-attribute
sighted: bool instance-attribute
sightedAt: int instance-attribute
ignoreSimilarity: bool instance-attribute
InputBulkUpdateObservable

Bases: InputUpdateObservable

ids: List[str] instance-attribute

observable_type

InputObservableTypeRequired

Bases: TypedDict

name: str instance-attribute
InputObservableType

Bases: InputObservableTypeRequired

isAttachment: bool instance-attribute
OutputObservableTypeRequired

Bases: TypedDict

name: str instance-attribute
isAttachment: bool instance-attribute
OutputObservableType

organisation

Bases: TypedDict

linkType: str instance-attribute
otherLinkType: str instance-attribute

Bases: TypedDict

toOrganisation: str instance-attribute
linkType: str instance-attribute
otherLinkType: str instance-attribute
OutputSharingProfile

Bases: TypedDict

name: str instance-attribute
description: str instance-attribute
autoShare: bool instance-attribute
editable: bool instance-attribute
permissionProfile: str instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute
InputOrganisationRequired

Bases: TypedDict

name: str instance-attribute
description: str instance-attribute
InputOrganisation

Bases: InputOrganisationRequired

taskRule: str instance-attribute
observableRule: str instance-attribute
locked: bool instance-attribute
OutputOrganisationRequired

Bases: TypedDict

name: str instance-attribute
description: str instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute
locked: bool instance-attribute
extraData: dict instance-attribute
OutputOrganisation

Bases: OutputOrganisationRequired

avatar: str instance-attribute
InputUpdateOrganisation

Bases: TypedDict

name: str instance-attribute
description: str instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute
locked: bool instance-attribute
avatar: str instance-attribute

page

InputCasePageRequired

Bases: TypedDict

title: str instance-attribute
content: str instance-attribute
category: str instance-attribute
InputCasePage

Bases: InputCasePageRequired

order: int instance-attribute
OutputCasePageRequired

Bases: TypedDict

id: str instance-attribute
createdBy: str instance-attribute
createdAt: int instance-attribute
title: str instance-attribute
content: str instance-attribute
slug: str instance-attribute
order: int instance-attribute
category: str instance-attribute
OutputCasePage

Bases: OutputCasePageRequired

updatedBy: str instance-attribute
updatedAt: int instance-attribute
InputUpdateCasePage

Bases: TypedDict

title: str instance-attribute
content: str instance-attribute
category: str instance-attribute
order: int instance-attribute

procedure

InputProcedureRequired

Bases: TypedDict

occurDate: int instance-attribute
patternId: str instance-attribute
InputProcedure

Bases: InputProcedureRequired

tactic: str instance-attribute
description: str instance-attribute
OutputProcedureRequired

Bases: TypedDict

occurDate: int instance-attribute
tactic: str instance-attribute
tacticLabel: str instance-attribute
extraData: dict instance-attribute
OutputProcedure

Bases: OutputProcedureRequired

description: str instance-attribute
patternId: str instance-attribute
patternName: str instance-attribute
InputUpdateProcedure

Bases: TypedDict

description: str instance-attribute
occurDate: int instance-attribute

profile

InputProfileRequired

Bases: TypedDict

name: str instance-attribute
InputProfile

Bases: InputProfileRequired

permissions: List[str] instance-attribute
OutputProfileRequired

Bases: TypedDict

name: str instance-attribute
editable: bool instance-attribute
isAdmin: bool instance-attribute
OutputProfile

Bases: OutputProfileRequired

permissions: List[str] instance-attribute
InputUpdateProfile

Bases: TypedDict

name: str instance-attribute
permissions: List[str] instance-attribute

share

OutputShareRequired

Bases: TypedDict

caseId: str instance-attribute
profileName: str instance-attribute
organisationName: str instance-attribute
owner: bool instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute
OutputShare
InputShareRequired

Bases: TypedDict

organisation: str instance-attribute
InputShare

Bases: InputShareRequired

share: bool instance-attribute
profile: str instance-attribute
taskRule: str instance-attribute
observableRule: str instance-attribute

task

InputTaskRequired

Bases: TypedDict

title: str instance-attribute
InputTask

Bases: InputTaskRequired

group: str instance-attribute
description: str instance-attribute
status: str instance-attribute
flag: bool instance-attribute
startDate: int instance-attribute
endDate: int instance-attribute
order: int instance-attribute
dueDate: int instance-attribute
assignee: str instance-attribute
mandatory: bool instance-attribute
OutputTaskRequired

Bases: TypedDict

title: str instance-attribute
group: str instance-attribute
status: str instance-attribute
flag: bool instance-attribute
order: int instance-attribute
mandatory: bool instance-attribute
extraData: dict instance-attribute
OutputTask

Bases: OutputTaskRequired

description: str instance-attribute
startDate: int instance-attribute
endDate: int instance-attribute
assignee: str instance-attribute
dueDate: int instance-attribute
InputUpdateTask

Bases: TypedDict

title: str instance-attribute
group: str instance-attribute
description: str instance-attribute
status: str instance-attribute
flag: bool instance-attribute
startDate: int instance-attribute
endDate: int instance-attribute
order: int instance-attribute
dueDate: int instance-attribute
assignee: str instance-attribute
mandatory: bool instance-attribute
InputBulkUpdateTask

Bases: InputUpdateTask

ids: List[str] instance-attribute

task_log

InputTaskLogRequired

Bases: TypedDict

message: str instance-attribute
InputTaskLog

Bases: InputTaskLogRequired

startDate: int instance-attribute
includeInTimeline: int instance-attribute
OutputTaskLogRequired

Bases: TypedDict

message: str instance-attribute
date: int instance-attribute
owner: str instance-attribute
extraData: dict instance-attribute
OutputTaskLog

Bases: OutputTaskLogRequired

attachments: List[dict] instance-attribute
includeInTimeline: int instance-attribute
InputUpdateTaskLog

Bases: TypedDict

message: str instance-attribute
includeInTimeline: int instance-attribute

timeline

OutputTimelineEventRequired

Bases: TypedDict

date: int instance-attribute
kind: str instance-attribute
entity: str instance-attribute
entityId: str instance-attribute
details: dict instance-attribute
OutputTimelineEvent

Bases: OutputTimelineEventRequired

endDate: int instance-attribute
OutputTimeline

Bases: TypedDict

events: List[OutputTimelineEvent] instance-attribute
InputCustomEventRequired

Bases: TypedDict

date: int instance-attribute
title: str instance-attribute
InputCustomEvent

Bases: InputCustomEventRequired

endDate: int instance-attribute
description: str instance-attribute
OutputCustomEventRequired

Bases: TypedDict

date: int instance-attribute
title: str instance-attribute
OutputCustomEvent

Bases: OutputCustomEventRequired

endDate: int instance-attribute
description: str instance-attribute
InputUpdateCustomEvent

Bases: TypedDict

date: int instance-attribute
endDate: int instance-attribute
title: str instance-attribute
description: str instance-attribute

user

InputUserRequired

Bases: TypedDict

login: str instance-attribute
name: str instance-attribute
profile: str instance-attribute
InputUser

Bases: InputUserRequired

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

Bases: TypedDict

organisationId: str instance-attribute
organisation: str instance-attribute
profile: str instance-attribute
OutputUserRequired

Bases: TypedDict

login: str instance-attribute
name: str instance-attribute
hasKey: bool instance-attribute
hasPassword: bool instance-attribute
hasMFA: bool instance-attribute
locked: bool instance-attribute
profile: str instance-attribute
organisation: str instance-attribute
type: str instance-attribute
extraData: dict instance-attribute
OutputUser

Bases: OutputUserRequired

email: str instance-attribute
permissions: List[str] instance-attribute
avatar: str instance-attribute
organisations: List[OutputOrganisationProfile] instance-attribute
defaultOrganisation: str instance-attribute
InputUpdateUser

Bases: TypedDict

name: str instance-attribute
organisation: str instance-attribute
profile: str instance-attribute
locked: bool instance-attribute
avatar: str instance-attribute
email: str instance-attribute
defaultOrganisation: str instance-attribute
InputUserOrganisationRequired

Bases: TypedDict

organisation: str instance-attribute
profile: str instance-attribute
InputUserOrganisation

Bases: InputUserOrganisationRequired

default: bool instance-attribute
OutputUserOrganisation

Bases: TypedDict

organisation: str instance-attribute
profile: str instance-attribute
default: bool instance-attribute

query

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

filters

FilterExpr = _Union['_FilterBase', dict] module-attribute
Lt(field, value)

Bases: _FilterBase

Field less than value.

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

Bases: _FilterBase

Field greater than value.

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

Bases: _FilterBase

Field less than or equal value.

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

Bases: _FilterBase

Field less than or equal value.

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

Bases: _FilterBase

Field not equal value.

Source code in thehive4py/query/filters.py
64
65
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
71
72
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
78
79
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
85
86
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
92
93
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
 99
100
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
106
107
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
113
114
def __init__(self, field: str):
    super().__init__(_contains=field)
Like(field, value)

Bases: _FilterBase

Field contains the value.

Source code in thehive4py/query/filters.py
120
121
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
127
128
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)
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