Paging

Author: dannywillems

README.md

@@ -42,6 +42,18 @@ def example_get_service_filter_plugin():
     assert all((i.tags == ["ntlm"] for i in response.json()))
 
 
+def example_get_service_filter_plugin_with_pagination():
+    """
+    Filter by fields. In this example, we want to have the NTLM services.
+    A list of plugins can be found in leakix.plugin
+    """
+    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
+    response = CLIENT.get_service(queries=[query_http_ntlm], page=1)
+    assert response.status_code() == 200
+    # check we only get NTML related services
+    assert all((i.tags == ["ntlm"] for i in response.json()))
+
+
 def example_get_leaks_filter_multiple_plugins():
     query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
     query_country = MustQuery(field=CountryField("France"))

example/example_client.py

@@ -27,6 +27,19 @@ def example_get_service_filter_plugin():
     assert all((i.tags == ["ntlm"] for i in response.json()))
 
 
+def example_get_service_filter_plugin_with_pagination():
+    """
+    Filter by fields. In this example, we want to have the NTLM services.
+    A list of plugins can be found in leakix.plugin.
+    Ask for page 1 (starts at 0)
+    """
+    query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
+    response = CLIENT.get_service(queries=[query_http_ntlm], page=1)
+    assert response.status_code() == 200
+    # check we only get NTML related services
+    assert all((i.tags == ["ntlm"] for i in response.json()))
+
+
 def example_get_leaks_filter_multiple_plugins():
     query_http_ntlm = MustQuery(field=PluginField(Plugin.HttpNTLM))
     query_country = MustQuery(field=CountryField("France"))
@@ -76,6 +89,7 @@ def example_get_plugins():
 if __name__ == "__main__":
     example_get_host_filter_plugin()
     example_get_service_filter_plugin()
+    example_get_service_filter_plugin_with_pagination()
     example_get_leaks_filter_multiple_plugins()
     example_get_leaks_multiple_filter_plugins_must_not()
     example_get_leak_raw_query()

leakix/client.py

@@ -25,6 +25,8 @@ class HostResult(Model):
 
 
 class Client:
+    MAX_RESULTS_PER_PAGE = 20
+
     def __init__(
         self,
         api_key: Optional[str] = None,
@@ -53,27 +55,31 @@ def __get(self, url, params):
         else:
             return ErrorResponse(response=r, response_json=r.json())
 
-    def get(self, scope: Scope, queries: Optional[List[Query]] = None):
+    def get(self, scope: Scope, queries: Optional[List[Query]] = None, page: int = 0):
+        if page < 0:
+            raise ValueError("Page argument must be a positive integer")
         if queries is None or len(queries) == 0:
             serialized_query = EmptyQuery().serialize()
         else:
             serialized_query = [q.serialize() for q in queries]
             serialized_query = " ".join(serialized_query)
             serialized_query = "%s" % serialized_query
         url = "%s/search" % self.base_url
-        r = self.__get(url=url, params={"scope": scope.value, "q": serialized_query})
+        r = self.__get(
+            url=url, params={"scope": scope.value, "q": serialized_query, "page": page}
+        )
         return r
 
-    def get_service(self, queries: Optional[List[Query]] = None):
-        r = self.get(Scope.SERVICE, queries=queries)
+    def get_service(self, queries: Optional[List[Query]] = None, page: int = 0):
+        r = self.get(Scope.SERVICE, queries=queries, page=page)
         if r.is_success():
             r.response_json = [
                 l9format.L9Event.from_dict(res) for res in r.response_json
             ]
         return r
 
-    def get_leak(self, queries: Optional[List[Query]] = None):
-        r = self.get(Scope.LEAK, queries=queries)
+    def get_leak(self, queries: Optional[List[Query]] = None, page: int = 0):
+        r = self.get(Scope.LEAK, queries=queries, page=page)
         if r.is_success():
             r.response_json = [
                 l9format.L9Event.from_dict(res) for res in r.response_json