Merge pull request #196 from DomainTools/IDEV-2477

[IDEV-2477] added irisQL support in python wrapper

Author: briluza

.github/workflows/test-build-publish.yml

@@ -45,8 +45,7 @@ jobs:
       - name: Install dependencies
         run: |
           python -m pip install --upgrade pip
-          pip install -e .
-          pip install -r ./requirements/development.txt
+          pip install -e ".[test]"
 
       - name: Setup E2E environment
         run: |

README.md

@@ -145,6 +145,32 @@ You can get the status code of a response outside of exception handling by doing
 api.domain_profile('google.com').status == 200
 ```
 
+IrisQL
+===================
+
+IrisQL is a query language for Iris Investigate that lets you express complex, multi-field searches in a single request. Pass the query as a raw string via the `irisql` parameter. The query must begin with `# IrisQL-1.0`.
+
+```python
+query = """# IrisQL-1.0
+DOMAIN CONTAINS "phishing"
+AND
+RISK_SCORE GREATER_THAN 85
+"""
+
+results = api.iris_investigate(irisql=query)
+print(results["results_count"])
+for domain in results:
+    print(domain["domain"])
+```
+
+Pagination parameters (`page_size`, `sort_by`, `position`) are supported alongside IrisQL via `**kwargs`:
+
+```python
+results = api.iris_investigate(irisql=query, page_size=50, sort_by="risk_score", position=0)
+```
+
+When `irisql` is set, any domain or filter parameters passed alongside it are silently ignored. IrisQL uses header-based authentication (`X-Api-Key`) automatically.
+
 Using the API Asynchronously
 ===================
 
@@ -201,6 +227,18 @@ Optionally, you can specify the desired format (html, xml, json, or list) of the
 domaintools domain_search google --max_length 10 -u $TEST_USER -k $TEST_KEY -f html
 ```
 
+IrisQL queries are supported via the `--irisql` flag on `iris_investigate`. The query must begin with `# IrisQL-1.0` on its own line:
+
+```bash
+domaintools iris_investigate --irisql $'# IrisQL-1.0\nDOMAIN CONTAINS "phishing"' -u $TEST_USER -k $TEST_KEY
+```
+
+Pagination parameters can be passed alongside the IrisQL query:
+
+```bash
+domaintools iris_investigate --irisql $'# IrisQL-1.0\nDOMAIN CONTAINS "phishing"' --page-size 50 --sort-by risk_score -u $TEST_USER -k $TEST_KEY
+```
+
 To avoid having to type in your API key repeatedly, you can specify them in `~/.dtapi` separated by a new line:
 
 ```bash
@@ -289,12 +327,11 @@ To add more e2e tests, put these in the `../tests/e2e` folder.
         source venv/bin/activate
     ```
 
-- Install dependencies.
+- Install dependencies (with test extras):
     ```bash
-        pip install -r requirements/development.txt
+        pip install -e ".[test]"
     ```
-
-- From the python_api project root directory, install the package.
+    Or without test dependencies:
     ```bash
         pip install -e .
     ```

domaintools/api.py

@@ -662,13 +662,14 @@ def iris_investigate(
         updated_after=None,
         include_domains_with_missing_field=None,
         exclude_domains_with_missing_field=None,
+        irisql=None,
         **kwargs,
     ):
         """Returns back a list of domains based on the provided filters.
 
         You can loop over results of your investigation as if it was a native Python list:
 
-            for result in api.iris_investigate(ip='199.30.228.112'):  # Enables looping over all related results
+            for result in api.iris_investigate(ip='199.30.228.112'):
 
         api.iris_investigate(QUERY)['results_count'] Returns the number of results returned with this request
         api.iris_investigate(QUERY)['total_count'] Returns the number of results available within Iris
@@ -677,9 +678,27 @@ def iris_investigate(
         api.iris_investigate(QUERY)['position'] Returns the position key that can be used to retrieve the next page:
             next_page = api.iris_investigate(QUERY, position=api.iris_investigate(QUERY)['position'])
 
-        for enrichment in api.iris_enrich(i):  # Enables looping over all returned enriched domains
+        IrisQL mode (mutually exclusive with all other search parameters):
+
+        irisql: str: A raw IrisQL query string. Must begin with '# IrisQL-1.0'.
+            Sent as a raw POST body (text/plain). When set, all domain/filter params are ignored.
+            Pagination params (page_size, sort_by, position) are still supported via **kwargs.
+
+            Example:
+                api.iris_investigate(irisql='# IrisQL-1.0\\nDOMAIN CONTAINS "phishing"', page_size=50, sort_by='risk_score')
 
         """
+        if irisql is not None:
+            if domains:
+                print("Warning: irisql is set — ignoring 'domains' and other search parameters. IrisQL query takes precedence.")
+            return self._results(
+                "iris-investigate",
+                "/v1/iris-investigate/",
+                items_path=("results",),
+                irisql=irisql,
+                **kwargs,
+            )
+
         # We put search_hash in the signature definition so the CLI can see it as a valid arg
         if search_hash:
             kwargs["search_hash"] = search_hash

domaintools/base_results.py

@@ -108,6 +108,17 @@ def _make_request(self):
                 "iris-enrich",
                 "iris-detect-escalate-domains",
             ]:
+                if self.product == "iris-investigate" and "irisql" in self.kwargs:
+                    irisql_query = self.kwargs["irisql"]
+                    auth_keys = {"api_username", "timestamp", "signature", "api_key"}
+                    query_params = {k: v for k, v in self.kwargs.items() if k != "irisql" and k not in auth_keys}
+                    query_params.update(self.api.extra_request_params)
+                    return session.post(
+                        url=self.url,
+                        content=irisql_query,
+                        params=query_params,
+                        headers={**headers, "Content-Type": "text/plain", "X-Api-Key": self.api.key},
+                    )
                 post_data = self.kwargs.copy()
                 post_data.update(self.api.extra_request_params)
                 return session.post(url=self.url, data=post_data, headers=headers)
@@ -277,6 +288,8 @@ def __exit__(self, *args):
 
     @property
     def json(self):
+        if self._data is not None:
+            return self
         self.kwargs.pop("format", None)
         return self.__class__(
             format="json",

domaintools/cli/api.py

@@ -230,28 +230,21 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
                     description=f"Preparing results with format of {response_format}...",
                 )
 
+                if name not in ("available_api_calls",) and not getattr(response, "product", None) in RTTF_PRODUCTS_LIST:
+                    response.data()
+
                 output = cls._get_formatted_output(
                     cmd_name=name, response=response, out_format=response_format
                 )
 
-                if isinstance(out_file, _io.TextIOWrapper):
-                    progress.update(
-                        task_id,
-                        description=f"Printing the results with format of {response_format}...",
-                    )
-                    # use rich `print` command to prettify the ouput in sys.stdout
-                    if name not in ("available_api_calls",) and response.product in RTTF_PRODUCTS_LIST:
-                        for feeds in response.response():
-                            print(feeds)
-                    else:
-                        print(response)
+            if isinstance(out_file, _io.TextIOWrapper):
+                if name not in ("available_api_calls",) and response.product in RTTF_PRODUCTS_LIST:
+                    for feeds in response.response():
+                        print(feeds)
                 else:
-                    progress.update(
-                        task_id,
-                        description=f"Writing results to {out_file}",
-                    )
-                    # if it's a file then write
-                    out_file.write(output if output.endswith("\n") else output + "\n")
+                    print(output)
+            else:
+                out_file.write(output if output.endswith("\n") else output + "\n")
         except Exception as e:
             if isinstance(e, ServiceException):
                 code = typer.style(getattr(e, "code", 400), fg=typer.colors.BRIGHT_RED)

domaintools/cli/commands/iris.py

@@ -23,6 +23,11 @@ def iris_investigate(
     create_date: str = typer.Option(None, "--create-date", help="The create date."),
     active: bool = typer.Option(None, "--active", help="The domains that are in active state"),
     search_hash: str = typer.Option(None, "--search-hash", help="The search hash to use"),
+    irisql: str = typer.Option(
+        None,
+        "--irisql",
+        help="IrisQL query string (must begin with '# IrisQL-1.0'). Mutually exclusive with domain/filter params. Pagination kwargs (--page-size, --sort-by, --position) are still supported.",
+    ),
     src_file: str = typer.Option(
         None,
         "-s",

domaintools_async/__init__.py

@@ -46,9 +46,21 @@ async def _make_async_request(self, session):
         session_params_and_headers = self._get_session_params_and_headers()
         headers = session_params_and_headers.get("headers")
         if self.product in ["iris-investigate", "iris-enrich", "iris-detect-escalate-domains"]:
-            post_data = self.kwargs.copy()
-            post_data.update(self.api.extra_request_params)
-            results = await session.post(url=self.url, data=post_data, headers=headers)
+            if self.product == "iris-investigate" and "irisql" in self.kwargs:
+                irisql_query = self.kwargs["irisql"]
+                auth_keys = {"api_username", "timestamp", "signature", "api_key"}
+                query_params = {k: v for k, v in self.kwargs.items() if k != "irisql" and k not in auth_keys}
+                query_params.update(self.api.extra_request_params)
+                results = await session.post(
+                    url=self.url,
+                    content=irisql_query,
+                    params=query_params,
+                    headers={**headers, "Content-Type": "text/plain", "X-Api-Key": self.api.key},
+                )
+            else:
+                post_data = self.kwargs.copy()
+                post_data.update(self.api.extra_request_params)
+                results = await session.post(url=self.url, data=post_data, headers=headers)
         elif self.product in ["iris-detect-manage-watchlist-domains"]:
             patch_data = self.kwargs.copy()
             patch_data.update(self.api.extra_request_params, headers=headers)

pyproject.toml

@@ -41,7 +41,7 @@ Homepage = "https://github.com/domaintools/python_api"
 domaintools = "domaintools.cli:run"
 
 [project.optional-dependencies]
-test = ["pytest", "mock"]
+test = ["pytest", "mock", "vcrpy", "pytest-asyncio", "pytest-cov", "yarl"]
 
 [tool.setuptools]
 packages = [
@@ -58,5 +58,8 @@ domaintools = ["specs/*.yaml"]
 [tool.setuptools.dynamic]
 version = { "file" = "VERSION" }
 
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+
 [tool.black]
 line-length = 110