WIP: #113 & #114

.gitignore

@@ -152,3 +152,11 @@ dmypy.json
 cython_debug/
 .DS_Store
 helpers/__pycache__/api.cpython-37.pyc
+
+# Python virtual environments
+pyvenv.cfg
+bin/
+lib/
+lib64/
+include/
+share/

docs/applications.md

@@ -4,6 +4,7 @@ The following methods call Veracode REST APIs and return JSON.
 
 ## Applications
 
+- `Applications().yield_all(policy_check_after(opt))` : get a list of Veracode applications. If provided, returns only applications that have a policy check date on or after `policy_check_after` (format is `yyyy-mm-dd`).
 - `Applications().get_all(policy_check_after(opt))` : get a list of Veracode applications (JSON format). If provided, returns only applications that have a policy check date on or after `policy_check_after` (format is `yyyy-mm-dd`).
 - `Applications().get(guid(opt),legacy_id(opt))`: get information for a single Veracode application using either the `guid` or the `legacy_id` (integer).
 - `Applications().get_by_name(name)`: get list of applications whose names contain the search string `name`.

requirements.txt

@@ -2,4 +2,5 @@ requests==2.32.4
 veracode-api-signing>=24.11.0
 Pygments>= 2.9.0
 idna>=3.7
-certifi>=2024.7.4
+certifi>=2024.7.4
+dacite>=1.8.0

veracode_api_py/apihelper.py

@@ -1,21 +1,21 @@
 # apihelper.py - API class for making network calls
-
-import requests
 import logging
-import json
 import time
-from requests.adapters import HTTPAdapter
+from datetime import datetime
+from typing import Iterable, TypeVar, Type
 
-from veracode_api_signing.exceptions import VeracodeAPISigningException
-from veracode_api_signing.plugin_requests import RequestsAuthPluginVeracodeHMAC
+import requests
+from dacite import from_dict, Config
+from requests.adapters import HTTPAdapter
 from veracode_api_signing.credentials import get_credentials
+from veracode_api_signing.plugin_requests import RequestsAuthPluginVeracodeHMAC
 from veracode_api_signing.regions import get_region_for_api_credential
 
-from .exceptions import VeracodeAPIError
-from .log import VeracodeLog as vlog
 from .constants import Constants
+from .exceptions import VeracodeAPIError
 
 logger = logging.getLogger(__name__)
+T = TypeVar('T')
 
 
 class APIHelper():
@@ -151,6 +151,41 @@ def _rest_paged_request(self, uri, method, element, params=None,fullresponse=Fal
         else:
             return all_data
 
+    def _yield_paginated_request(self, uri, method, entity: Type[T], params=None) -> Iterable[T]:
+        """Paginates the request and yields the entities.
+
+        Args:
+            uri: The URI to request.
+            method: The HTTP method to use.
+            entity: The entity to yield.
+            params: The parameters to pass to the request.
+            fullresponse: Whether to return the full response.
+
+        Returns:
+            An iterable of the entities.
+        """
+
+        element = entity._element
+
+        page = 0
+        more_pages = True
+
+        while more_pages:
+            params['page'] = page
+            page_data = self._rest_request(uri, method, params)
+            total_pages = page_data.get('page', {}).get('total_pages', 0)
+            data_page = page_data.get('_embedded', {}).get(element, [])
+            for data in data_page:
+                config = Config(
+                    type_hooks={
+                        datetime: lambda x: datetime.fromisoformat(x.replace('Z', '+00:00')) if isinstance(x, str) else x
+                    }
+                )
+                yield from_dict(data_class=entity, data=data, config=config)
+
+            page += 1
+            more_pages = page < total_pages
+
     def _xml_request(self, url, method, params=None, files=None):
         # base request method for XML APIs, handles what little error handling there is around these APIs
         if method not in ["GET", "POST"]:
@@ -160,14 +195,14 @@ def _xml_request(self, url, method, params=None, files=None):
             session = requests.Session()
             session.mount(self.baseurl, HTTPAdapter(max_retries=3))
             request = requests.Request(method, url, params=params, files=files,
-                                       auth=RequestsAuthPluginVeracodeHMAC(), headers=self._prepare_headers(method,'xml'))
+                                       auth=RequestsAuthPluginVeracodeHMAC(), headers=self._prepare_headers(method, 'xml'))
             prepared_request = request.prepare()
             r = session.send(prepared_request)
             if 200 <= r.status_code <= 299:
                 if r.status_code == 204:
                     # retry after wait
                     time.sleep(self.retry_seconds)
-                    return self._xml_request(url,method,params)
+                    return self._xml_request(url, method, params)
                 elif r.content is None:
                     logger.debug("HTTP response body empty:\r\n{}\r\n{}\r\n{}\r\n\r\n{}\r\n{}\r\n{}\r\n"
                                  .format(r.request.url, r.request.headers, r.request.body, r.status_code, r.headers,

veracode_api_py/applications.py

@@ -1,13 +1,24 @@
 #applications.py - API class for Applications API calls
 
 import json
+from typing import Iterable
 from urllib import parse
 from uuid import UUID
 
 from .apihelper import APIHelper
 from .constants import Constants
+from .models.applications_entity import ApplicationsEntity
+
 
 class Applications():    
+    def yield_all(self, policy_check_after=None) -> Iterable[ApplicationsEntity]:
+        if policy_check_after == None:
+            params={}
+        else:
+            params={"policy_compliance_checked_after": policy_check_after}
+
+        return APIHelper()._yield_paginated_request('appsec/v1/applications',"GET", ApplicationsEntity, params=params)
+
     def get_all(self,policy_check_after=None):
         if policy_check_after == None:
             params={}

veracode_api_py/findings.py

@@ -1,14 +1,40 @@
 #findings.py - API class for Findings API and related calls
 
 import json
+from typing import Iterable
 from uuid import UUID
 
+from veracode_api_py.models.findings_entity import FindingsEntity
+
 from .apihelper import APIHelper
 
 LINE_NUMBER_SLOP = 3 #adjust to allow for line number movement  
 
 class Findings():
 
+    def yield_all(self, app: UUID, scantype='STATIC', annot='TRUE', request_params=None, sandbox: UUID = None) -> Iterable[FindingsEntity]:
+        if request_params == None:
+            request_params = {}
+
+        scantypes = ""
+        scantype = scantype.split(',')
+        for st in scantype:
+            if st in ['STATIC', 'DYNAMIC', 'MANUAL', 'SCA']:
+                if len(scantypes) > 0:
+                    scantypes += ","
+                scantypes += st
+        if len(scantypes) > 0:
+            request_params['scan_type'] = scantypes
+        # note that scantype='ALL' will result in no scan_type parameter as in API
+
+        request_params['include_annot'] = annot
+
+        if sandbox != None:
+            request_params['context'] = sandbox
+
+        uri = "appsec/v2/applications/{}/findings".format(app)
+        return APIHelper()._yield_paginated_request(uri, "GET", FindingsEntity, request_params)
+
     def get_findings(self,app: UUID,scantype='STATIC',annot='TRUE',request_params=None,sandbox: UUID=None):
         #Gets a list of  findings for app using the Veracode Findings API
         if request_params == None:

veracode_api_py/models/applications_entity.py

@@ -0,0 +1,91 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import List, Optional
+
+
+@dataclass
+class Scan:
+    scan_type: Optional[str]
+    status: Optional[str]
+    modified_date: Optional[datetime]
+    scan_url: Optional[str]
+    internal_status: Optional[str]
+
+
+@dataclass
+class BusinessUnit:
+    id: Optional[int]
+    name: Optional[str]
+    guid: Optional[str]
+
+
+@dataclass
+class BusinessOwner:
+    name: Optional[str]
+    email: Optional[str]
+
+
+@dataclass
+class Policy:
+    guid: Optional[str]
+    name: Optional[str]
+    is_default: Optional[bool]
+    policy_compliance_status: Optional[str]
+
+
+@dataclass
+class Team:
+    team_id: Optional[int]
+    team_name: Optional[str]
+    guid: Optional[str]
+
+
+@dataclass
+class CustomField:
+    name: Optional[str]
+    value: Optional[str]
+
+
+@dataclass
+class Settings:
+    nextday_consultation_allowed: Optional[bool]
+    static_scan_xpa_or_dpa: Optional[bool]
+    dynamic_scan_approval_not_required: Optional[bool]
+    sca_enabled: Optional[bool]
+    static_scan_xpp_enabled: Optional[bool]
+
+
+@dataclass
+class Profile:
+    name: Optional[str]
+    tags: Optional[str]
+    business_unit: Optional[BusinessUnit]
+    business_owners: Optional[List[BusinessOwner]]
+    archer_app_name: Optional[str]
+    enterprise_id: Optional[int]
+    policies: Optional[List[Policy]]
+    teams: Optional[List[Team]]
+    custom_fields: Optional[List[CustomField]]
+    description: Optional[str]
+    settings: Optional[Settings]
+    git_repo_url: Optional[str]
+    vendor_rescan: Optional[bool]
+    business_criticality: Optional[str]
+
+
+@dataclass
+class ApplicationsEntity:
+    id: Optional[int]
+    oid: Optional[int]
+    last_completed_scan_date: Optional[datetime]
+    guid: Optional[str]
+    created: Optional[datetime]
+    modified: Optional[datetime]
+    alt_org_id: Optional[int]
+    app_profile_url: Optional[str]
+    scans: Optional[List[Scan]]
+    last_policy_compliance_check_date: Optional[datetime]
+    profile: Optional[Profile]
+    results_url: Optional[str]
+
+    _element: str = 'applications'

veracode_api_py/models/findings_entity.py

@@ -0,0 +1,59 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+
+
+@dataclass
+class CWE:
+    id: Optional[int]
+    name: Optional[str]
+    href: Optional[str]
+
+
+@dataclass
+class FindingCategory:
+    id: Optional[int]
+    name: Optional[str]
+    href: Optional[str]
+
+
+@dataclass
+class FindingDetails:
+    severity: Optional[int]
+    cwe: Optional[CWE]
+    file_path: Optional[str]
+    file_name: Optional[str]
+    module: Optional[str]
+    relative_location: Optional[int]
+    finding_category: Optional[FindingCategory]
+    procedure: Optional[str]
+    exploitability: Optional[int]
+    attack_vector: Optional[str]
+    file_line_number: Optional[int]
+
+
+@dataclass
+class FindingStatus:
+    first_found_date: Optional[datetime]
+    status: Optional[str]
+    resolution: Optional[str]
+    mitigation_review_status: Optional[str]
+    new: Optional[bool]
+    resolution_status: Optional[str]
+    last_seen_date: Optional[datetime]
+
+
+@dataclass
+class FindingsEntity:
+    issue_id: Optional[int]
+    scan_type: Optional[str]
+    description: Optional[str]
+    count: Optional[int]
+    context_type: Optional[str]
+    context_guid: Optional[str]
+    violates_policy: Optional[bool]
+    finding_status: Optional[FindingStatus]
+    finding_details: Optional[FindingDetails]
+    build_id: Optional[int]
+
+    _element: str = 'findings'