Skip to content

ptnad.api.BQLAPI 

BQLAPI(client)

Methods:

Name Description
execute

Execute a BQL query on a specific source.
execute_raw

Execute a BQL query on a specific source and return the full response.
Source code in src/ptnad/api/bql.py 
24
25
def __init__(self, client) -> None:
    self.client = client

execute 

execute(query: str, source: str = '2') -> Any

Execute a BQL query on a specific source.

Args: query (str): The BQL query to execute. source (str): The identifier of the storage to query. Defaults to "2" (live).

Returns: Any: The result of the query execution.

Raises: PTNADAPIError: If there's an error executing the query.

Source code in src/ptnad/api/bql.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def execute(self, query: str, source: str = "2") -> Any:
    """
    Execute a BQL query on a specific source.

    Args:
        query (str): The BQL query to execute.
        source (str): The identifier of the storage to query. Defaults to "2" (live).

    Returns:
        Any: The result of the query execution.

    Raises:
        PTNADAPIError: If there's an error executing the query.

    """
    try:
        response = self._send_query(query, source)
        return response["result"]
    except PTNADAPIError as e:
        e.operation = "execute BQL query"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to execute BQL query: {str(e)}")

execute_raw 

execute_raw(query: str, source: str = '2') -> BQLResponse

Execute a BQL query on a specific source and return the full response.

Args: query (str): The BQL query to execute. source (str): The identifier of the storage to query. Defaults to "2" (live).

Returns: BQLResponse: An object containing the query results, execution time, total hits, and debug info.

Raises: PTNADAPIError: If there's an error executing the query.

Source code in src/ptnad/api/bql.py 
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
def execute_raw(self, query: str, source: str = "2") -> BQLResponse:
    """
    Execute a BQL query on a specific source and return the full response.

    Args:
        query (str): The BQL query to execute.
        source (str): The identifier of the storage to query. Defaults to "2" (live).

    Returns:
        BQLResponse: An object containing the query results, execution time, total hits, and debug info.

    Raises:
        PTNADAPIError: If there's an error executing the query.

    """
    try:
        response = self._send_query(query, source)
        return BQLResponse(
            result=response["result"],
            took=response["took"],
            total=response["total"],
            debug=response.get("debug")
        )
    except PTNADAPIError as e:
        e.operation = "execute BQL query"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to execute BQL query: {str(e)}")

ptnad.api.MonitoringAPI 

MonitoringAPI(client)

Methods:

Name Description
get_active_triggers

Get all active triggers (triggers with status other than 'green').
get_status

Get the current status.
get_trigger_by_id

Get a specific trigger by its ID.
get_triggers

Get the list of triggers.
get_triggers_by_type

Get all triggers of a specific type.
Source code in src/ptnad/api/monitoring.py 
55
56
def __init__(self, client) -> None:
    self.client = client

get_active_triggers 

get_active_triggers() -> List[Trigger]

Get all active triggers (triggers with status other than 'green').

Returns: List[Trigger]: A list of active Trigger objects.

Raises: PTNADAPIError: If there's an error retrieving the triggers.

Source code in src/ptnad/api/monitoring.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
def get_active_triggers(self) -> List[Trigger]:
    """
    Get all active triggers (triggers with status other than 'green').

    Returns:
        List[Trigger]: A list of active Trigger objects.

    Raises:
        PTNADAPIError: If there's an error retrieving the triggers.

    """
    triggers = self.get_triggers()
    return [trigger for trigger in triggers if trigger.status != "green"]

get_status 

get_status() -> MonitoringStatus

Get the current status.

Returns: MonitoringStatus: An object containing the current status and any problems.

Raises: PTNADAPIError: If there's an error retrieving the status.

Source code in src/ptnad/api/monitoring.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def get_status(self) -> MonitoringStatus:
    """
    Get the current status.

    Returns:
        MonitoringStatus: An object containing the current status and any problems.

    Raises:
        PTNADAPIError: If there's an error retrieving the status.

    """
    try:
        response = self.client.get("/monitoring/status").json()
        return MonitoringStatus(**response)
    except PTNADAPIError as e:
        e.operation = "get monitoring status"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get monitoring status: {str(e)}")

get_trigger_by_id 

get_trigger_by_id(trigger_id: str) -> Trigger | None

Get a specific trigger by its ID.

Args: trigger_id (str): The ID of the trigger to retrieve.

Returns: Optional[Trigger]: The Trigger object if found, None otherwise.

Raises: PTNADAPIError: If there's an error retrieving the trigger.

Source code in src/ptnad/api/monitoring.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def get_trigger_by_id(self, trigger_id: str) -> Trigger | None:
    """
    Get a specific trigger by its ID.

    Args:
        trigger_id (str): The ID of the trigger to retrieve.

    Returns:
        Optional[Trigger]: The Trigger object if found, None otherwise.

    Raises:
        PTNADAPIError: If there's an error retrieving the trigger.

    """
    triggers = self.get_triggers()
    return next((trigger for trigger in triggers if trigger.id == trigger_id), None)

get_triggers 

get_triggers() -> List[Trigger]

Get the list of triggers.

Returns: List[Trigger]: A list of Trigger objects.

Raises: PTNADAPIError: If there's an error retrieving the triggers.

Source code in src/ptnad/api/monitoring.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get_triggers(self) -> List[Trigger]:
    """
    Get the list of triggers.

    Returns:
        List[Trigger]: A list of Trigger objects.

    Raises:
        PTNADAPIError: If there's an error retrieving the triggers.

    """
    try:
        response = self.client.get("/monitoring/triggers").json()
        return [Trigger(**trigger) for trigger in response.get("results", [])]
    except PTNADAPIError as e:
        e.operation = "get triggers"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get triggers: {str(e)}")

get_triggers_by_type 

get_triggers_by_type(trigger_type: str) -> List[Trigger]

Get all triggers of a specific type.

Args: trigger_type (str): The type of triggers to retrieve.

Returns: List[Trigger]: A list of Trigger objects of the specified type.

Raises: PTNADAPIError: If there's an error retrieving the triggers.

Source code in src/ptnad/api/monitoring.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get_triggers_by_type(self, trigger_type: str) -> List[Trigger]:
    """
    Get all triggers of a specific type.

    Args:
        trigger_type (str): The type of triggers to retrieve.

    Returns:
        List[Trigger]: A list of Trigger objects of the specified type.

    Raises:
        PTNADAPIError: If there's an error retrieving the triggers.

    """
    triggers = self.get_triggers()
    return [trigger for trigger in triggers if trigger.type == trigger_type]

ptnad.api.RepListsAPI 

RepListsAPI(client)

Methods:

Name Description
add_dynamic_list_item

Add an item to a dynamic reputation list.
bulk_add_items

Add multiple items to a dynamic reputation list.
bulk_delete_items

Delete multiple items from a dynamic reputation list.
create_list

Create a new reputation list.
delete_list

Delete a reputation list.
get_all_lists

Get all reputation lists using pagination.
get_dynamic_list_items

Get items from a dynamic reputation list.
get_list

Get information about a specific reputation list.
get_lists

Get a list of reputation lists.
get_stats

Get statistics about reputation lists.
remove_item

Remove an item from a dynamic reputation list.
update_list

Update a reputation list.
Source code in src/ptnad/api/replists.py 
8
9
def __init__(self, client) -> None:
    self.client = client

add_dynamic_list_item 

add_dynamic_list_item(external_key: str, value: str, attributes: Dict[str, Any] | None = None) -> Dict[str, Any]

Add an item to a dynamic reputation list.

Args: external_key (str): External key of the reputation list. value (str): Value to add to the list. attributes (Optional[Dict[str, Any]]): Additional attributes for the item.

Returns: Dict[str, Any]: Information about the added item.

Raises: PTNADAPIError: If there's an error adding the item.

Source code in src/ptnad/api/replists.py 
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def add_dynamic_list_item(self, external_key: str, value: str, attributes: Dict[str, Any] | None = None) -> Dict[str, Any]:
    """
    Add an item to a dynamic reputation list.

    Args:
        external_key (str): External key of the reputation list.
        value (str): Value to add to the list.
        attributes (Optional[Dict[str, Any]]): Additional attributes for the item.

    Returns:
        Dict[str, Any]: Information about the added item.

    Raises:
        PTNADAPIError: If there's an error adding the item.

    """
    try:
        response = self.client.post(f"/replists/dynamic/{external_key}/{value}", json=attributes or {})
        if response.status_code in (200, 201):
            return response.json()
        raise PTNADAPIError(f"Failed to add item to reputation list {external_key}: {response.text}")
    except PTNADAPIError as e:
        e.operation = f"add item to reputation list {external_key}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to add item to reputation list {external_key}: {str(e)}")

bulk_add_items 

bulk_add_items(external_key: str, items: List[Dict[str, Any]]) -> None

Add multiple items to a dynamic reputation list.

Args: external_key (str): External key of the reputation list. items (List[Dict[str, Any]]): List of items to add, each item should have 'value' and optionally 'attrs'.

Raises: PTNADAPIError: If there's an error adding the items.

Source code in src/ptnad/api/replists.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
def bulk_add_items(self, external_key: str, items: List[Dict[str, Any]]) -> None:
    """
    Add multiple items to a dynamic reputation list.

    Args:
        external_key (str): External key of the reputation list.
        items (List[Dict[str, Any]]): List of items to add, each item should have 'value' and optionally 'attrs'.

    Raises:
        PTNADAPIError: If there's an error adding the items.

    """
    try:
        self.client.post(f"/replists/dynamic/{external_key}/_bulk", json=items)
    except PTNADAPIError as e:
        e.operation = f"bulk add items to reputation list {external_key}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to bulk add items to reputation list {external_key}: {str(e)}")

bulk_delete_items 

bulk_delete_items(external_key: str, values: List[str]) -> None

Delete multiple items from a dynamic reputation list.

Args: external_key (str): External key of the reputation list. values (List[str]): List of values to delete from the list.

Raises: PTNADAPIError: If there's an error deleting the items.

Source code in src/ptnad/api/replists.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def bulk_delete_items(self, external_key: str, values: List[str]) -> None:
    """
    Delete multiple items from a dynamic reputation list.

    Args:
        external_key (str): External key of the reputation list.
        values (List[str]): List of values to delete from the list.

    Raises:
        PTNADAPIError: If there's an error deleting the items.

    """
    try:
        response = self.client.post(f"/replists/dynamic/{external_key}/_delete", json=values)
        if response.status_code != 204:
            raise PTNADAPIError(f"Failed to bulk delete items from reputation list {external_key}: {response.text}")
    except PTNADAPIError as e:
        e.operation = f"bulk delete items from reputation list {external_key}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to bulk delete items from reputation list {external_key}: {str(e)}")

create_list 

create_list(name: str, type: str, color: str, description: str | None = None, content: Union[str, List[str], None] = None, external_key: str | None = None) -> Dict[str, Any]

Create a new reputation list.

Args: name (str): Name of the reputation list. Must be a valid slug (letters, numbers, underscores, hyphens). type (str): Type of the reputation list ('ip', 'dn', 'uri', or 'md5'). color (str): Color code for the reputation list ('0' to '7'). description (Optional[str]): Description of the reputation list. content (Optional[Union[str, List[str]]]): Content of the reputation list. Can be a string or a list of strings that will be joined with newlines. external_key (Optional[str]): External key for the reputation list.

Returns: Dict[str, Any]: Information about the created reputation list.

Raises: ValidationError: If the input parameters are invalid. PTNADAPIError: If there's an error creating the list.

Source code in src/ptnad/api/replists.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def create_list(self, name: str, type: str, color: str, description: str | None = None,
                content: Union[str, List[str], None] = None, external_key: str | None = None) -> Dict[str, Any]:
    """
    Create a new reputation list.

    Args:
        name (str): Name of the reputation list. Must be a valid slug (letters, numbers, underscores, hyphens).
        type (str): Type of the reputation list ('ip', 'dn', 'uri', or 'md5').
        color (str): Color code for the reputation list ('0' to '7').
        description (Optional[str]): Description of the reputation list.
        content (Optional[Union[str, List[str]]]): Content of the reputation list. Can be a string or a list of strings that will be joined with newlines.
        external_key (Optional[str]): External key for the reputation list.

    Returns:
        Dict[str, Any]: Information about the created reputation list.

    Raises:
        ValidationError: If the input parameters are invalid.
        PTNADAPIError: If there's an error creating the list.

    """
    if not self._is_valid_slug(name):
        raise ValidationError("Name must be a valid slug consisting of letters, numbers, underscores or hyphens.")

    data = {
        "name": name,
        "type": type,
        "color": color,
    }
    if description:
        data["description"] = description
    if content:
        # Convert list of strings to newline-separated string if needed
        if isinstance(content, list):
            data["content"] = "\n".join(content)
        else:
            data["content"] = content
    if external_key:
        data["external_key"] = external_key

    try:
        response = self.client.post("/replists", json=data)
        if response.status_code == 201:
            return response.json()
        error_message = response.json() if response.headers.get("Content-Type") == "application/json" else response.text
        raise PTNADAPIError(f"Failed to create reputation list. Status code: {response.status_code}. Error: {error_message}")
    except PTNADAPIError as e:
        e.operation = "create reputation list"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to create reputation list: {str(e)}")

delete_list 

delete_list(list_id: int) -> None

Delete a reputation list.

Args: list_id (int): ID of the reputation list to delete.

Raises: PTNADAPIError: If there's an error deleting the reputation list.

Source code in src/ptnad/api/replists.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def delete_list(self, list_id: int) -> None:
    """
    Delete a reputation list.

    Args:
        list_id (int): ID of the reputation list to delete.

    Raises:
        PTNADAPIError: If there's an error deleting the reputation list.

    """
    try:
        response = self.client.delete(f"/replists/{list_id}")
        if response.status_code != 204:
            raise PTNADAPIError(f"Failed to delete reputation list {list_id}: {response.text}")
    except PTNADAPIError as e:
        e.operation = f"delete reputation list {list_id}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to delete reputation list {list_id}: {str(e)}")

get_all_lists 

get_all_lists(search: str | None = None, ordering: str | None = None, limit: int = 100) -> List[Dict[str, Any]]

Get all reputation lists using pagination.

Args: search (Optional[str]): Keyword to filter the lists by name, description, or vendor name. ordering (Optional[str]): Field to sort the results by (id, name, color, type, created, modified, items_count, description, vendor__name). Use prefix with '-' for descending order. limit (int): Number of lists to fetch per request (default: 100).

Returns: List[Dict[str, Any]]: A list of dictionaries containing all reputation list information.

Raises: PTNADAPIError: If there's an error retrieving the lists.

Source code in src/ptnad/api/replists.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def get_all_lists(self, search: str | None = None, ordering: str | None = None,
                  limit: int = 100) -> List[Dict[str, Any]]:
    """
    Get all reputation lists using pagination.

    Args:
        search (Optional[str]): Keyword to filter the lists by name, description, or vendor name.
        ordering (Optional[str]): Field to sort the results by (id, name, color, type, created, modified, items_count, description, vendor__name). Use prefix with '-' for descending order.
        limit (int): Number of lists to fetch per request (default: 100).

    Returns:
        List[Dict[str, Any]]: A list of dictionaries containing all reputation list information.

    Raises:
        PTNADAPIError: If there's an error retrieving the lists.

    """
    all_lists = []
    offset = 0

    while True:
        response = self._get_lists_data(search, ordering, limit, offset)
        lists = response["results"]
        all_lists.extend(lists)

        if response["next"] is None:
            break

        offset += limit

    return all_lists

get_dynamic_list_items 

get_dynamic_list_items(external_key: str, ordering: str | None = None) -> List[Dict[str, Any]]

Get items from a dynamic reputation list.

Args: external_key (str): External key of the reputation list. ordering (Optional[str]): Field to sort the results by (value or modified).

Returns: List[Dict[str, Any]]: List of items in the reputation list.

Raises: PTNADAPIError: If there's an error retrieving the items.

Source code in src/ptnad/api/replists.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
def get_dynamic_list_items(self, external_key: str, ordering: str | None = None) -> List[Dict[str, Any]]:
    """
    Get items from a dynamic reputation list.

    Args:
        external_key (str): External key of the reputation list.
        ordering (Optional[str]): Field to sort the results by (value or modified).

    Returns:
        List[Dict[str, Any]]: List of items in the reputation list.

    Raises:
        PTNADAPIError: If there's an error retrieving the items.

    """
    params = {}
    if ordering:
        params["ordering"] = ordering

    try:
        response = self.client.get(f"/replists/dynamic/{external_key}", params=params).json()
        return response["results"]
    except PTNADAPIError as e:
        e.operation = f"get items from reputation list {external_key}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get items from reputation list {external_key}: {str(e)}")

get_list 

get_list(list_id: int) -> Dict[str, Any]

Get information about a specific reputation list.

Args: list_id (int): ID of the reputation list.

Returns: Dict[str, Any]: Information about the reputation list.

Raises: PTNADAPIError: If there's an error retrieving the list.

Source code in src/ptnad/api/replists.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def get_list(self, list_id: int) -> Dict[str, Any]:
    """
    Get information about a specific reputation list.

    Args:
        list_id (int): ID of the reputation list.

    Returns:
        Dict[str, Any]: Information about the reputation list.

    Raises:
        PTNADAPIError: If there's an error retrieving the list.

    """
    try:
        response = self.client.get(f"/replists/{list_id}").json()
        return response
    except PTNADAPIError as e:
        e.operation = f"get reputation list {list_id}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get reputation list {list_id}: {str(e)}")

get_lists 

get_lists(search: str | None = None, ordering: str | None = None, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]

Get a list of reputation lists.

Args: search (Optional[str]): Keyword to filter the lists by name, description, or vendor name. ordering (Optional[str]): Field to sort the results by (id, name, color, type, created, modified, items_count, description, vendor__name). Use prefix '-' for descending order. limit (int): Maximum number of lists to return (default: 100). offset (int): Number of lists to skip (default: 0).

Returns: List[Dict[str, Any]]: A list of dictionaries containing reputation list information.

Raises: PTNADAPIError: If there's an error retrieving the lists.

Source code in src/ptnad/api/replists.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_lists(self, search: str | None = None, ordering: str | None = None,
              limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]:
    """
    Get a list of reputation lists.

    Args:
        search (Optional[str]): Keyword to filter the lists by name, description, or vendor name.
        ordering (Optional[str]): Field to sort the results by (id, name, color, type, created, modified, items_count, description, vendor__name). Use prefix '-' for descending order.
        limit (int): Maximum number of lists to return (default: 100).
        offset (int): Number of lists to skip (default: 0).

    Returns:
        List[Dict[str, Any]]: A list of dictionaries containing reputation list information.

    Raises:
        PTNADAPIError: If there's an error retrieving the lists.

    """
    response = self._get_lists_data(search, ordering, limit, offset)
    return response["results"]

get_stats 

get_stats() -> Dict[str, Any]

Get statistics about reputation lists.

Returns: Dict[str, Any]: Statistics about reputation lists.

Raises: PTNADAPIError: If there's an error retrieving the statistics.

Source code in src/ptnad/api/replists.py 
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def get_stats(self) -> Dict[str, Any]:
    """
    Get statistics about reputation lists.

    Returns:
        Dict[str, Any]: Statistics about reputation lists.

    Raises:
        PTNADAPIError: If there's an error retrieving the statistics.

    """
    try:
        response = self.client.get("/replists/stats").json()
        return response
    except PTNADAPIError as e:
        e.operation = "get reputation lists statistics"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get reputation lists statistics: {str(e)}")

remove_item 

remove_item(external_key: str, value: str) -> None

Remove an item from a dynamic reputation list.

Args: external_key (str): External key of the reputation list. value (str): Value to remove from the reputation list.

Raises: PTNADAPIError: If there's an error removing the item.

Source code in src/ptnad/api/replists.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
def remove_item(self, external_key: str, value: str) -> None:
    """
    Remove an item from a dynamic reputation list.

    Args:
        external_key (str): External key of the reputation list.
        value (str): Value to remove from the reputation list.

    Raises:
        PTNADAPIError: If there's an error removing the item.

    """
    try:
        response = self.client.delete(f"/replists/dynamic/{external_key}/{value}")
        if response.status_code != 204:
            raise PTNADAPIError(f"Failed to remove item from reputation list {external_key}: {response.text}")
    except PTNADAPIError as e:
        e.operation = f"remove item from reputation list {external_key}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to remove item from reputation list {external_key}: {str(e)}")

update_list 

update_list(list_id: int, **kwargs) -> Dict[str, Any]

Update a reputation list.

Args: list_id (int): ID of the reputation list to update. **kwargs: Fields to update (color, name, type, description, content).

Returns: Dict[str, Any]: Updated information about the reputation list.

Raises: PTNADAPIError: If there's an error updating the list.

Source code in src/ptnad/api/replists.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def update_list(self, list_id: int, **kwargs) -> Dict[str, Any]:
    """
    Update a reputation list.

    Args:
        list_id (int): ID of the reputation list to update.
        **kwargs: Fields to update (color, name, type, description, content).

    Returns:
        Dict[str, Any]: Updated information about the reputation list.

    Raises:
        PTNADAPIError: If there's an error updating the list.

    """
    try:
        response = self.client.patch(f"/replists/{list_id}", json=kwargs).json()
        return response
    except PTNADAPIError as e:
        e.operation = f"update reputation list {list_id}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to update reputation list {list_id}: {str(e)}")

ptnad.api.SignaturesAPI 

SignaturesAPI(client)

Methods:

Name Description
apply_changes

Apply changes made to Rules and commit them to sensors.
get_classes

Get a list of signature classes.
get_rule

Get information about a specific Rule.
get_rules

Get a list of Rules.
get_stats

Get statistics about Rules.
revert_changes

Revert changes made to Rules.
update_rule

Update a Rule.
Source code in src/ptnad/api/signatures.py 
7
8
def __init__(self, client) -> None:
    self.client = client

apply_changes 

apply_changes() -> Dict[str, str]

Apply changes made to Rules and commit them to sensors.

Returns: Dict[str, str]: A dictionary with the hashsum of the package.

Raises: PTNADAPIError: If there's an error applying the changes.

Source code in src/ptnad/api/signatures.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def apply_changes(self) -> Dict[str, str]:
    """
    Apply changes made to Rules and commit them to sensors.

    Returns:
        Dict[str, str]: A dictionary with the hashsum of the package.

    Raises:
        PTNADAPIError: If there's an error applying the changes.

    """
    try:
        response = self.client.post("/signatures/commit").json()
        if "hashsum" in response:
            return response
        if "fatal_error" in response or "other_errors" in response:
            errors = []
            if response.get("fatal_error"):
                errors.append(response["fatal_error"])
            if response.get("other_errors"):
                errors.extend(response["other_errors"])
            raise PTNADAPIError(f"Failed to commit signature changes: {', '.join(errors)}")
        return response
    except PTNADAPIError as e:
        e.operation = f"commit signature changes: {', '.join(errors)}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to apply Rule changes: {str(e)}")

get_classes 

get_classes(search: str | None = None, ordering: str | None = None, **filters) -> List[Dict[str, Any]]

Get a list of signature classes.

Args: search (Optional[str]): Keyword to filter the classes. ordering (Optional[str]): Field to sort the results by. **filters: Additional filters (name, title, priority).

Returns: List[Dict[str, Any]]: A list of dictionaries containing signature class information.

Raises: PTNADAPIError: If there's an error retrieving the classes.

Source code in src/ptnad/api/signatures.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def get_classes(self, search: str | None = None, ordering: str | None = None, **filters) -> List[Dict[str, Any]]:
    """
    Get a list of signature classes.

    Args:
        search (Optional[str]): Keyword to filter the classes.
        ordering (Optional[str]): Field to sort the results by.
        **filters: Additional filters (name, title, priority).

    Returns:
        List[Dict[str, Any]]: A list of dictionaries containing signature class information.

    Raises:
        PTNADAPIError: If there's an error retrieving the classes.

    """
    params = {k: v for k, v in filters.items() if v is not None}
    if search:
        params["search"] = search
    if ordering:
        params["ordering"] = ordering

    try:
        response = self.client.get("/signatures/classes", params=params).json()
        return response["results"]
    except PTNADAPIError as e:
        e.operation = "get signature classes"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get signature classes: {str(e)}")

get_rule 

get_rule(rule_id: int) -> Dict[str, Any]

Get information about a specific Rule.

Args: rule_id (int): sid of the Rule.

Returns: Dict[str, Any]: Information about the Rule.

Raises: PTNADAPIError: If there's an error retrieving the rule.

Source code in src/ptnad/api/signatures.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def get_rule(self, rule_id: int) -> Dict[str, Any]:
    """
    Get information about a specific Rule.

    Args:
        rule_id (int): sid of the Rule.

    Returns:
        Dict[str, Any]: Information about the Rule.

    Raises:
        PTNADAPIError: If there's an error retrieving the rule.

    """
    try:
        response = self.client.get(f"/signatures/rules/{rule_id}").json()
        return response
    except PTNADAPIError as e:
        e.operation = f"get Rule {rule_id}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get rule {rule_id}: {str(e)}")

get_rules 

get_rules(search: str | None = None, ordering: str | None = None, limit: int = 100, offset: int = 0, **filters) -> List[Dict[str, Any]]

Get a list of Rules.

Args: search (Optional[str]): Keyword to filter the rules. ordering (Optional[str]): Field to sort the results by. limit (int): Maximum number of rules to return (default: 100). offset (int): Number of rules to skip (default: 0). *filters: Additional filters. Available filters: sid: Filter by sid (can be a single value or a list) sid__gte: sid greater than or equal to sid__lt: sid less than vendor: Filter by vendor name (can be a single value or a list) enabled: Filter by enabled status (true or false) has_redef: Filter by has_redef status (true or false) has_exceptions: Filter by has_exceptions status (true or false) priority: Filter by priority (if priority=4, finds all rules with priority>=4) cls: Filter by class name (can be a single value or a list) diff: Filter by rule changes, valid values (can be a list): added (+), updated (), removed (-), unchanged (=). Available in 12.2+ has_error: Filter by has_error status (true or false) client: Search for IP address in src_adr and dst_adr server: Search for IP address in src_adr and dst_adr

Returns: List[Dict[str, Any]]: A list of dictionaries containing Rule information.

Raises: PTNADAPIError: If there's an error retrieving the rules.

Source code in src/ptnad/api/signatures.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def get_rules(self, search: str | None = None, ordering: str | None = None,
              limit: int = 100, offset: int = 0, **filters) -> List[Dict[str, Any]]:
    """
    Get a list of Rules.

    Args:
        search (Optional[str]): Keyword to filter the rules.
        ordering (Optional[str]): Field to sort the results by.
        limit (int): Maximum number of rules to return (default: 100).
        offset (int): Number of rules to skip (default: 0).
        **filters: Additional filters. Available filters:
            sid: Filter by sid (can be a single value or a list)
            sid__gte: sid greater than or equal to
            sid__lt: sid less than
            vendor: Filter by vendor name (can be a single value or a list)
            enabled: Filter by enabled status (true or false)
            has_redef: Filter by has_redef status (true or false)
            has_exceptions: Filter by has_exceptions status (true or false)
            priority: Filter by priority (if priority=4, finds all rules with priority>=4)
            cls: Filter by class name (can be a single value or a list)
            diff: Filter by rule changes, valid values (can be a list): added (+), updated (*), removed (-), unchanged (=). Available in 12.2+
            has_error: Filter by has_error status (true or false)
            client: Search for IP address in src_adr and dst_adr
            server: Search for IP address in src_adr and dst_adr

    Returns:
        List[Dict[str, Any]]: A list of dictionaries containing Rule information.

    Raises:
        PTNADAPIError: If there's an error retrieving the rules.

    """
    response = self._get_rules_data(search, ordering, limit, offset, **filters)
    return response["results"]

get_stats 

get_stats() -> Dict[str, Any]

Get statistics about Rules.

Returns: Dict[str, Any]: Statistics about Rules.

Raises: PTNADAPIError: If there's an error retrieving the statistics.

Source code in src/ptnad/api/signatures.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def get_stats(self) -> Dict[str, Any]:
    """
    Get statistics about Rules.

    Returns:
        Dict[str, Any]: Statistics about Rules.

    Raises:
        PTNADAPIError: If there's an error retrieving the statistics.

    """
    try:
        response = self.client.get("/signatures/stats").json()
        return response
    except PTNADAPIError as e:
        e.operation = "get Rules statistics"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to get Rules statistics: {str(e)}")

revert_changes 

revert_changes() -> None

Revert changes made to Rules.

Raises: PTNADAPIError: If there's an error reverting the changes.

Source code in src/ptnad/api/signatures.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def revert_changes(self) -> None:
    """
    Revert changes made to Rules.

    Raises:
        PTNADAPIError: If there's an error reverting the changes.

    """
    try:
        self.client.post("/signatures/rollback")
    except PTNADAPIError as e:
        e.operation = "revert Rule changes"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to revert Rule changes: {str(e)}")

update_rule 

update_rule(rule_id: int, **kwargs) -> Dict[str, Any]

Update a Rule.

Args: rule_id (int): ID of the Rule to update. **kwargs: Fields to update (enabled, action, msg, etc.).

Returns: Dict[str, Any]: Updated information about the Rule.

Raises: PTNADAPIError: If there's an error updating the rule.

Source code in src/ptnad/api/signatures.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def update_rule(self, rule_id: int, **kwargs) -> Dict[str, Any]:
    """
    Update a Rule.

    Args:
        rule_id (int): ID of the Rule to update.
        **kwargs: Fields to update (enabled, action, msg, etc.).

    Returns:
        Dict[str, Any]: Updated information about the Rule.

    Raises:
        PTNADAPIError: If there's an error updating the rule.

    """
    try:
        response = self.client.patch(f"/signatures/rules/{rule_id}", json=kwargs).json()
        return response
    except PTNADAPIError as e:
        e.operation = f"update Rule {rule_id}"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to update Rule {rule_id}: {str(e)}")