Merge pull request #146 from intezer/feat/query-by-url

feature(analyses-history) - add url and scanned_url filters to histor… TKT-5316

CHANGES

@@ -1,3 +1,7 @@
+1.21.5
+_______
+- Add url and scanned_url to url analyses history filters
+
 1.21.4
 _______
 - Always refresh access token on 'unauthorized' http status

intezer_sdk/__init__.py

@@ -1 +1 @@
-__version__ = '1.21.4'
+__version__ = '1.21.5'

intezer_sdk/analyses_history.py

@@ -112,6 +112,8 @@ def query_url_analyses_history(*,
                                sub_verdicts: List[str] = None,
                                did_download_file: bool = None,
                                submitted_url: str = None,
+                               scanned_url: str = None,
+                               url: str = None,
                                aggregated_view: bool = False,
                                limit: int = DEFAULT_LIMIT,
                                offset: int = DEFAULT_OFFSET
@@ -127,6 +129,8 @@ def query_url_analyses_history(*,
     :param sub_verdicts: Filter by the analysis's verdict
     :param did_download_file: Should the result be aggregated by latest url.
     :param submitted_url: Filter by specific url
+    :param scanned_url: Filter by specific scanned_url
+    :param url: Filter by submitted url or scanned url
     :param aggregated_view: Should the result be aggregated by latest url.
     :param limit: Number of analyses returned by the query.
     :param offset: Number of analyses to skips the before beginning to return the analyses.
@@ -148,6 +152,10 @@ def query_url_analyses_history(*,
         filters['did_download_file'] = did_download_file
     if submitted_url:
         filters['submitted_url'] = submitted_url
+    if scanned_url:
+        filters['scanned_url'] = scanned_url
+    if url:
+        filters['url'] = url
     if sub_verdicts:
         filters['sub_verdicts'] = sub_verdicts
 

intezer_sdk/analysis.py

@@ -401,18 +401,13 @@ def from_latest_analysis(cls,
         now = datetime.datetime.now()
         yesterday = now - datetime.timedelta(days=days_threshold_for_latest_analysis)
 
+        url = _clean_url(url) if exact_match else url
         analysis_history_url_result = query_url_analyses_history(start_date=yesterday,
                                                                  end_date=now,
+                                                                 url=url,
                                                                  aggregated_view=True,
                                                                  api=api)
-        all_analyses_reports = analysis_history_url_result.all()
-
-
-        analyses_ids = [report['analysis_id'] for report in all_analyses_reports
-                        if url in (report.get('scanned_url'), report.get('submitted_url'))]
-        if not analyses_ids and not exact_match:
-            analyses_ids = [report['analysis_id'] for report in all_analyses_reports
-                            if _clean_url(url) in (_clean_url(report.get('scanned_url', '')), _clean_url(report.get('submitted_url', '')))]
+        analyses_ids = [report['analysis_id'] for report in analysis_history_url_result.all()]
 
         if not analyses_ids:
             return None

test_requirements.txt

@@ -1,4 +1,4 @@
 # This is not used by the project, but is used by the CI/CD pipeline to install dependencies, update setup.py for package dependencies.
 requests>=2.29.0,<3
 responses==0.25.0
-pytest==8.0.1
+pytest==8.1.1

tests/unit/test_results.py

@@ -190,7 +190,7 @@ def test_url_analyses_history_happy_flow(self):
                 verdicts=['malicious'],
                 sub_verdicts=['phishing'],
                 did_download_file=True,
-                submitted_url='https://example_trusted.com'
+                url='https://example_trusted.com'
             )
             for result in results:
                 assert result