Datapath/pyhuntress
2
0
Fork 0
mirror of https://github.com/brygphilomena/pyhuntress.git synced 2025-11-04 00:07:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

made pagination work

Browse Source
This commit is contained in:
Peter Annabel 2025-07-24 18:26:17 -05:00
parent 146431e818
commit e2bb030b71
27 changed files with 101 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/pyhuntress/clients/huntress_client.py
View File

@ -37,16 +37,16 @@ class HuntressClient(ABC):
def _get_url(self) -> str:
pass
def _make_request(
def _make_request( # noqa: C901
self,
method: RequestMethod,
url: str,
data: RequestData | None = None,
rawdata: RequestData | None = None,
# rawdata: RequestData | None = None,
params: RequestParams | None = None,
headers: dict[str, str] | None = None,
retry_count: int = 0,
stream: bool = False,
stream: bool = False, # noqa: FBT001, FBT002
) -> Response:
"""
Make an API request using the specified method, endpoint, data, and parameters.
@ -76,19 +76,19 @@ class HuntressClient(ABC):
method,
url,
headers=headers,
json=data,
params=cast(dict[str, Any], params or {}),
stream=stream,
)
elif rawdata:
response = requests.request(
method,
url,
headers=headers,
data=rawdata,
data=data,
params=cast(dict[str, Any], params or {}),
stream=stream,
)
# elif rawdata:
# response = requests.request(
# method,
# url,
# headers=headers,
# data=rawdata,
# params=cast(dict[str, Any], params or {}),
# stream=stream,
# )
else:
response = requests.request(
method,

2
src/pyhuntress/clients/managedsat_client.py
View File

@ -135,7 +135,7 @@ class HuntressSATAPIClient(HuntressClient):
auth_response = self._make_request(
"POST",
token_url,
rawdata={
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,

6
src/pyhuntress/endpoints/managedsat/AccountsEndpoint.py
View File

@ -52,10 +52,10 @@ class AccountsEndpoint(
PaginatedResponse[SATData]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATAccounts,

6
src/pyhuntress/endpoints/managedsat/AccountsIdAssignmentsEndpoint.py
View File

@ -38,10 +38,10 @@ class AccountsIdAssignmentsEndpoint(
PaginatedResponse[SATAssignments]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATAssignments,

6
src/pyhuntress/endpoints/managedsat/AccountsIdUsersEndpoint.py
View File

@ -38,10 +38,10 @@ class AccountsIdUsersEndpoint(
PaginatedResponse[SATUsers]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATUsers,

6
src/pyhuntress/endpoints/managedsat/AssignmentsEndpoint.py
View File

@ -52,10 +52,10 @@ class AssignmentsEndpoint(
PaginatedResponse[SATData]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATAssignments,

6
src/pyhuntress/endpoints/managedsat/AssignmentsIdEndpoint.py
View File

@ -49,10 +49,10 @@ class AssignmentsIdEndpoint(
PaginatedResponse[SATAssignments]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATAssignments,

6
src/pyhuntress/endpoints/managedsat/AssignmentsIdLearnerActivityEndpoint.py
View File

@ -38,10 +38,10 @@ class AssignmentsIdLearnerActivityEndpoint(
PaginatedResponse[SATLearnerActivities]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATLearnerActivities,

6
src/pyhuntress/endpoints/managedsat/AssignmentsIdLearnersEndpoint.py
View File

@ -38,10 +38,10 @@ class AssignmentsIdLearnersEndpoint(
PaginatedResponse[SATLearners]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATLearners,

6
src/pyhuntress/endpoints/managedsat/DepartmentsEndpoint.py
View File

@ -52,10 +52,10 @@ class DepartmentsEndpoint(
PaginatedResponse[SATDepartments]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATDepartments,

6
src/pyhuntress/endpoints/managedsat/DepartmentsIdEndpoint.py
View File

@ -38,10 +38,10 @@ class DepartmentsIdEndpoint(
PaginatedResponse[SATDepartments]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATDepartments,

6
src/pyhuntress/endpoints/managedsat/EpisodesEndpoint.py
View File

@ -52,10 +52,10 @@ class EpisodesEndpoint(
PaginatedResponse[SATData]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATEpisodes,

6
src/pyhuntress/endpoints/managedsat/EpisodesIdEndpoint.py
View File

@ -38,10 +38,10 @@ class EpisodesIdEndpoint(
PaginatedResponse[SATEpisodes]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATEpisodes,

6
src/pyhuntress/endpoints/managedsat/GroupsEndpoint.py
View File

@ -52,10 +52,10 @@ class GroupsEndpoint(
PaginatedResponse[SATGroups]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATGroups,

6
src/pyhuntress/endpoints/managedsat/GroupsIdEndpoint.py
View File

@ -38,10 +38,10 @@ class GroupsIdEndpoint(
PaginatedResponse[SATGroups]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATGroups,

6
src/pyhuntress/endpoints/managedsat/LearnersEndpoint.py
View File

@ -52,10 +52,10 @@ class LearnersEndpoint(
PaginatedResponse[SATLearners]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATLearners,

6
src/pyhuntress/endpoints/managedsat/LearnersIdEndpoint.py
View File

@ -38,10 +38,10 @@ class LearnersIdEndpoint(
PaginatedResponse[SATLearners]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATLearners,

6
src/pyhuntress/endpoints/managedsat/PhishingCampaignScenariosEndpoint.py
View File

@ -52,10 +52,10 @@ class PhishingCampaignScenariosEndpoint(
PaginatedResponse[SATData]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATPhishingScenarios,

6
src/pyhuntress/endpoints/managedsat/PhishingCampaignScenariosIdEndpoint.py
View File

@ -45,10 +45,10 @@ class PhishingCampaignScenariosIdEndpoint(
PaginatedResponse[SATPhishingScenarios]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATPhishingScenarios,

6
src/pyhuntress/endpoints/managedsat/PhishingCampaignsEndpoint.py
View File

@ -52,10 +52,10 @@ class PhishingCampaignsEndpoint(
PaginatedResponse[SATData]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATPhishingCampaigns,

6
src/pyhuntress/endpoints/managedsat/PhishingCampaignsIdEndpoint.py
View File

@ -43,10 +43,10 @@ class PhishingCampaignsIdEndpoint(
PaginatedResponse[SATPhishingCampaigns]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATPhishingCampaigns,

6
src/pyhuntress/endpoints/managedsat/PhishingScenariosEndpoint.py
View File

@ -52,10 +52,10 @@ class PhishingScenariosEndpoint(
PaginatedResponse[SATPhishingScenarios]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATPhishingScenarios,

6
src/pyhuntress/endpoints/managedsat/PhishingScenariosIdEndpoint.py
View File

@ -38,10 +38,10 @@ class PhishingScenariosIdEndpoint(
PaginatedResponse[SATPhishingScenarios]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATPhishingScenarios,

6
src/pyhuntress/endpoints/managedsat/UsersEndpoint.py
View File

@ -54,10 +54,10 @@ class UsersEndpoint(
PaginatedResponse[SATData]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATUsers,

6
src/pyhuntress/endpoints/managedsat/UsersIdEndpoint.py
View File

@ -39,10 +39,10 @@ class UsersIdEndpoint(
PaginatedResponse[SATUsers]: The initialized PaginatedResponse object.
"""
if params:
params["page"] = page
params["limit"] = limit
params["page[number]"] = page
params["page[size]"] = limit
else:
params = {"page": page, "limit": limit}
params = {"page[number]": page, "page[size]": limit}
return PaginatedResponse(
super()._make_request("GET", params=params),
SATUsers,

5
src/pyhuntress/responses/paginated_response.py
View File

@ -84,8 +84,11 @@ class PaginatedResponse(Generic[TModel]):
self.endpointmodel = endpointmodel
self.endpoint = endpoint
self.limit = limit
# The following for SIEM is in the response body, not the headers
# Get page data from the response body
try:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('pagination', {}))
except:
self.parsed_pagination_response = parse_response_body(json.loads(response.content.decode('utf-8')).get('meta.page', {}))
self.params = params
if self.parsed_pagination_response is not None:
# Huntress SIEM API gives us a handy response to parse for Pagination

12
src/pyhuntress/utils/helpers.py
View File

@ -77,12 +77,19 @@ def parse_response_body(
elif body.get("current_page") is not None:
if body.get("current_page") > 1:
result["prev_page"] = body.get("current_page") - 1
elif body.get("currentPage") is not None:
if body.get("currentPage") > 1:
result["prev_page"] = body.get("currentPage") - 1
if body.get("next_page") is not None:
result["next_page"] = body.get("next_page")
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
result["next_page"] = body.get("currentPage") + 1
if body.get("last_page") is not None:
result["last_page"] = body.get("last_page")
elif body.get("lastPage") is not None:
result["last_page"] = body.get("lastPage")
elif body.get("last_page") is None and body.get("current_page") is not None:
result["last_page"] = math.ceil(body.get("total_count")/body.get("limit"))
@ -92,12 +99,17 @@ def parse_response_body(
result["has_next_page"] = True
elif body.get("current_page") is not None and body.get("next_page") is None:
result["has_next_page"] = False
elif body.get("currentPage") is not None and body.get("currentPage") < body.get("lastPage"):
result["has_next_page"] = True
if body.get("has_prev_page"):
result["has_prev_page"] = body.get("has_prev_page")
elif body.get("current_page") is not None:
if body.get("current_page") > 1:
result["has_prev_page"] = True
elif body.get("currentPage") is not None:
if body.get("currentPage") > 1:
result["has_prev_page"] = True
return result