PolySwarm Customer API v3

$ pip install polyswarm-api

$ python setup.py install

from polyswarm_api.api import PolyswarmAPI

api_key = "1234123412341234123412341234"
community_name = "default"
api = PolyswarmAPI(key=api_key, community=community_name)

FILE = '/home/user/malicious.bin'

positives = 0
total = 0

instance = api.submit(FILE)
result = api.wait_for(instance)

if result.failed:
    print(f'Failed to get results')
    sys.exit()

print('Engine Assertions:')
for assertion in result.assertions:
    if assertion.verdict:
        positives += 1
    total += 1
    print('\tEngine {} asserts {}'.\
            format(assertion.author_name,
                   'Malicious' if assertion.verdict else 'Benign'))

print(f'Positives: {positives}')
print(f'Total: {total}')
print(f'PolyScore: {result.polyscore}\n')

print(f'sha256: {result.sha256}')
print(f'sha1: {result.sha1}')
print(f'md5: {result.md5}')
print(f'Extended type: {result.extended_type}')
print(f'First Seen: {result.first_seen}')
print(f'Last Seen: {result.last_seen}\n')

print(f'Permalink: {result.permalink}')

result = api.submit('./malicious-enc.zip',
                    preprocessing={'type': 'zip', 'password': 'infected'})
print(result.status)

URL = 'https://polyswarm.io'

positives = 0
total = 0

instance = api.submit(URL, artifact_type='url')
result = api.wait_for(instance)

if result.failed:
    print(f'Failed to get results')
    sys.exit()

print('Engine Assertions:')
for assertion in result.assertions:
    if assertion.verdict:
        positives += 1
    total += 1
    print('\tEngine {} asserts {}'.\
            format(assertion.author_name,
                   'Malicious' if assertion.verdict else 'Benign'))

print(f'Positives: {positives}')
print(f'Total: {total}\n')

print(f'Permalink: {result.permalink}')

instance = api.submit(URL, artifact_type='url', scan_config='most-time')
result = api.wait_for(instance)

instance = api.rescan("275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f")
result = api.wait_for(instance)

if result.failed:
    print(f'Failed to get results')
    sys.exit()

positives = 0
total = 0

print('Engine Assertions:')
for assertion in result.assertions:
    if assertion.verdict:
        positives += 1
    total += 1
    print('\tEngine {} asserts {}'.\
            format(assertion.author_name,
                   'Malicious' if assertion.verdict else 'Benign'))

print(f'Positives: {positives}')
print(f'Total: {total}')
print(f'PolyScore: {result.polyscore}\n')

print(f'sha256: {result.sha256}')
print(f'sha1: {result.sha1}')
print(f'md5: {result.md5}')
print(f'Extended type: {result.extended_type}')
print(f'First Seen: {result.first_seen}')
print(f'Last Seen: {result.last_seen}\n')

print(f'Permalink: {result.permalink}')

OUTPUT_DIR = '/tmp/'
EICAR_HASH = '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f'

artifact = api.download(OUTPUT_DIR, EICAR_HASH)

report = api.report_create(type='scan', format='pdf', instance_id=instance_id)
print(f'Report ID: {report.id} (State: {report.state})')

# `report_wait_for()` is a method to fetch the report progress over
# and over until is not in PENDING state anymore
report = api.report_wait_for(report.id, timeout=timeout_seconds)
if report.state == 'SUCCEEDED':
    response = requests.get(report.url, stream=True)
    response.raise_for_status()
    with open(f'scan-{instance_id}.pdf', 'wb') as f:
        response.raw.decode_content = True
        shutil.copyfileobj(response.raw, f)
else:
    print(f'Report failed (State: {report.state})')

report = api.report_create(type='sandbox_zip', format='zip', sandbox_task_id=123456, template_metadata={'zip_report_ids': [123], 'sandbox_artifact_types': ['report', 'raw_report', 'screenshot', 'jarm'] })

results = api.report_template_list()
for template in results:
   print(f'ID: {template.id}')
   print(f'Created: {template.created}')
   print(f'Name: {template.template_name}')
   print(f'Color: {template.primary_color}\n')

result = api.report_template_create(template_name='testreport1')

result = api.report_template_delete(98453877554394669)

result = api.report_template_get(95389624286242180)
print(f'ID: {result.id}')
print(f'Created: {result.created}')
print(f'Name: {result.template_name}')
print(f'Color: {result.primary_color}\n')

# sha256, md5, and sha1 supported
EICAR_HASH = '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f'

positives = 0
total = 0

try:
    results = api.search(EICAR_HASH)

    for result in results:
        if result.failed:
            print(f'Failed to get result.')
            break

        if not result.assertions:
            print('Artifact not scanned yet - Run rescan for Engine Assertions.')
        else:
            print('Engine Assertions:')

            for assertion in result.assertions:
                if assertion.verdict:
                    positives += 1
                total += 1
                print('\tEngine {} asserts {}'. \
                      format(assertion.author_name,
                             'Malicious' if assertion.verdict else 'Benign'))

        print(f'Positives: {positives}')
        print(f'Total: {total}')
        print(f'PolyScore: {result.polyscore}\n')

        print(f'sha256: {result.sha256}')
        print(f'sha1: {result.sha1}')
        print(f'md5: {result.md5}')
        print(f'Extended type: {result.extended_type}')
        print(f'First Seen: {result.first_seen}')
        print(f'Last Seen: {result.last_seen}\n')

        print(f'Permalink: {result.permalink}')
except exceptions.NoResultsException:
    print(f'No results for the provided hash.')

query = 'artifact.sha256:"275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f"'

results = api.search_by_metadata(query)

# Our query is by cryptographic hash; we expect at most 1 result.
# Regardless, it's good practice to properly handle multiple results.
for result in results:
    print(f"Artifact Attributes: {result.artifact}")

query = 'artifact.sha256:"275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f"'

results = api.search_by_metadata(query, exclude=["exiftool","lief","polyunite.malware_family"])

# Process results
for result in results:
    print(f"Artifact Attributes: {result.json}")  # Debugging: Print full response to verify fields   

query = 'artifact.sha256:"275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f"'

results = api.search_by_metadata(query, include=["pefile","polyunite.malware_family"])

# Process results
for result in results:
    print(f"Artifact Attributes: {result.json}")  # Debugging: Print full response to verify fields   

results = api.iocs_by_hash('sha256', 'aac08c6f7474c979acf2a3aef1f2727820ece755001530cdebf346b5d1ae2ccb', hide_known_good=True)
for result in results:
    iocs = result.json
    print(f"ips: {iocs['ips']}")
    print(f"urls: {iocs['urls']}")
    print(f"ttps: {iocs['ttps']}")
    print(f"imphash: {iocs['imphash']}")

# IOC Search by ip
for result in api.search_by_ioc(ip="108.159.227.121"):
    print(f"sha256: {result.json}")


# IOC Search by domain
for result in api.search_by_ioc(domain="img-s-msn-com.akamaized.net"):
    print(f"sha256: {result.json}")

# IOC Search by MITRE ttp
for result in api.search_by_ioc(ttp="T1060"):
    print(f"sha256: {result.json}")

results = api.check_known_hosts(domains=["polyswarm.network"], ips=["0.0.0.0"]):
for result in results:
    ioc = result.json
    print(f"type:{ioc.type}, host:{ioc.host}, source:{ioc.source}, good:{ioc.good}")

result = api.sandbox_file('./malicious.exe', 'triage', 'windows11-21h2-x64')
print(result.status)

result = api.sandbox_file('./malicious-enc.zip',
                          'triage',
                          'windows11-21h2-x64',
                          preprocessing={'type': 'zip', 'password': 'infected'},
                          network_enabled=True)
print(result.status)

result = api.sandbox_url('www.polyswarm.io', 'triage', 'windows11-21h2-x64', browser='edge')
print(result.status)

result = api.sandbox_url(None,
                         'cape',
                         'win-10-build-19041',
                         artifact='/path/to/qrcode.png',
                         preprocessing={'type': 'qrcode'},
                         browser='firefox')
print(result.status)

result = api.sandbox(42445563653708569, 'triage', 'windows11-21h2-x64', True)
print(result.status)

result = api.sandbox_file('./tests/eicar.yara', 'triage', 'windows11-21h2-x64')

sandboxes = api.sandbox_providers()
print(sandboxes)

task = api.sandbox_task_status(53445563653708569)
print(task)

task = api.download_id('./outdir', 53445563653708569)
print(task)

latest = api.sandbox_task_latest('18e5b8fe65e8f73c3a4a637c258c02aeec8a6ab702b15b7ee73f5631a9879e40', 'triage')
print(latest)

tasks = api.sandbox_my_tasks_list(sandbox='triage')
print(tasks)

tasks = api.sandbox_task_list(sandbox='triage', start_date='2023-10-31', status="SUCCEEDED")
print(tasks)

new_ruleset = api.ruleset_create(name='eicar',
                                 rules=open('eicar.yara').read(),
                                 description='eicar ruleset')
print(f'ID: {new_ruleset.id}')

rulesets = api.ruleset_list()

for ruleset in rulesets:
    print(f'ID: {ruleset.id}')

# updating the ruleset yara rules (can also update name and description)
api.ruleset_update(new_ruleset.id, rules=open('another.yara').read())

api.ruleset_delete(new_ruleset.id)

ruleset = api.ruleset_get(57989886451857569)

ruleset = api.live_start(ruleset.id)
print(f'ID: {ruleset.livescan_id}')

# reverse chronologically ordered iterator
results = api.live_feed(since=999999)
for result in results:
    print(f'ID: {result.id}')

# you can inspect more details about a single result
# based on its id, it also provides a download link
# to the file and the origial yara rule used
# these extra info does not come directly from the
# feed listing method for performance reasons
result = api.live_result(91163237970748480)
print(f'ID: {result.id}')
print(f'URL: {result.download_url}')

api.live_feed_delete([91163237970748480])

ruleset = api.ruleset_get(57989886451857569)
ruleset = api.live_stop(ruleset.id)

historical = api.historical_create(rule=open('eicar.yara').read())
print(f'ID: {historical.id}')

# the only update you can perform on a historical hunt
# is to cancel the hunt before it finishes
api.historical_update(49988514210960880)

# you can also list all historical hunts you have in your account
results = api.historical_list(since=9999999)
for result in results:
    print(f'ID: {result.id}')

# you can retrieve extra information about the hunt
# this also includes a consolidated results csv
historical = api.historical_get(48011760326110718)
print(f'ID: {historical.id}')
print(f'Results CSV: {historical.results_csv_uri}')

# you can check the results of a historical hunt
results = api.historical_results(48011760326110718)
for result in results:
    print(f'ID: {result.id}')

# retrieve a single result with extra information
result = api.historical_result(89734617019442134)
print(f'ID: {result.id}')
print(f'URL: {result.download_url}')

# delete an undesirable result
api.historical_results_delete([89734617019442134])

# you can delete a historical hunt
# keep in mind that this is an async process and the
# hunt will be scheduled for deletio..
api.historical_delete(49988514210960880)

SINCE = 60 # Fetch stream from the last 60 minutes
streams = api.stream(since=SINCE)

for stream in streams:
    print(f'ID: {stream.id}')
    print(f'URI: {stream.uri}')
    print(f'Created: {stream.created}')
    print(f'Community: {stream.community}')