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.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}")

Artifact Attributes: {'created': '2019-05-03T23:43:42.298385+00:00', 'id': '90465088441197206', 'md5': '44d88612fea8a8f36de82e1278abb02f', 'sha1': '3395856ce81f2b7382dee72602f798b642f14140', 'sha256': '275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f'}

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: {result.sha256}")

    # Artifacts that have not been scanned will lack the scan object in Metadata results.
    if not hasattr(result, 'scan'):
        print(f"Artifact has not been scanned.")
        continue

    # Engines choose whether to respond to each scan request.
    # We cannot assume any given Engine Result Object exists, so much check first.
    if not 'ClamAV' in result.scan['latest_scan']:
        print(f"ClamAV did not deliver an assertion for the latest scan of this Artifact.")
        continue

    # If the scan object exists, the scan.latest_scan object will also exist.
    print(f"ClamAV asserted: {result.scan['latest_scan']['ClamAV']['assertion']}")
    print(f"ClamAV identified the malware family as: {result.scan['latest_scan']['ClamAV']['metadata']['malware_family']}")

Artifact: 275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f
ClamAV asserted: malicious
ClamAV identified the malware family as: Win.Test.EICAR_HDB-1

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: {result.sha256}")

    # The hash Tool applies to all Artifacts and is eventually run on all Artifacts.
    # There is a window, however, between submission and Tool execution, so we verify
    # the output from the Tool exists before attempting to access output members.
    if not hasattr(result, 'hash'):
        print(f"The hash Tool hasn't been run on this Artifact yet.")
        continue

    print(f"ssdeep fuzzy hash of Artifact: {result.hash['ssdeep']}")

Artifact: 275a021bbfb6489e54d471899f7db9d1663fc695ec2fe2a2c4538aabf651fd0f
ssdeep fuzzy hash of Artifact: 3:a+JraNvsgzsVqSwHq9:tJuOgzsko

query = 'hash.sha1:"1aefc2ebaf1a78f23473ce6275b0b514bbcdfb08"'

results = api.search_by_metadata(query)

for result in results:
    print(f"Artifact Attributes: {result.artifact}")

<?xml version="1.0" encoding="utf-8"?>
<resources>
    ...
    <string name="MT_Bin_dup_0x7f0c0020">Android Telegram</string>
    <string name="MT_Bin_dup_0x7f0c0021">10000</string>
    <string name="MT_Bin_dup_0x7f0c0022">82.137.218.185</string>
    ...
</resources>

results = api.search_by_metadata("*", ip="82.137.218.185")

for result in results:
    print(f"Artifact Attributes: {result.artifact}")

query = 'hash.sha256:"0f73ac8839f153cf0e830554d9b34af2ea90fd6514ed3992b66a96bc9c12bb4b"'

results = api.search_by_metadata(query)

for result in results:
    print(f"Artifact Attributes: {result.artifact}")

{
    "artifact": {
        "created": "2020-03-10T10:16:50.900548+00:00",
        "id": "71592690635387748",
        "md5": "766e5ecf6b1d86abf401ad9223de857d",
        "sha1": "f1271aa0ccf79d16b036bac5320ed4349af69b65",
        "sha256": "0f73ac8839f153cf0e830554d9b34af2ea90fd6514ed3992b66a96bc9c12bb4b"
    },
    ...
    "strings": {
        "domains": [
            "V.mr",
            "",
            "covid-19-e9057.appspot.com",
            "p.to",
            "II1046766097017-4va56jc12ajt308tpbuge0tc5iqla179.apps.googleusercontent.com",
            "b.mc",
            "YJ.cz",
            "6.om",
            "6.gm",
            "covid-19-e9057.firebaseio.com"
        ],
        ...
    }
}

$ polyswarm --fmt sha256 search metadata 'strings.domains:covid-19-e9057*'

query = 'strings.urls:*2no.co*1O5aW'

results = api.search_by_metadata(query)

for result in results:
    print(f"Artifact Attributes: {result.artifact}")

query = 'scan.latest_scan.\*.metadata.malware_family:*cve*'

results = api.search_by_metadata(query)

for result in results:
    print(f"Artifact Attributes: {result.artifact}")

import os
from polyswarm_api.api import PolyswarmAPI
from datetime import datetime, timedelta

end = datetime.utcnow().replace(microsecond=0)

start = end - timedelta(minutes=5)

for metadata_result_obj in api.search_by_metadata('artifact.created:[{} TO {}]'.format(start.isoformat(), end.isoformat())):
    if metadata_result_obj.malicious and metadata_result_obj.malicious > 2:
        print("{} marked malicious by {} engines and benign by {}".format(metadata_result_obj.sha256, metadata_result_obj.malicious, metadata_result_obj.benign))
        artifact_instance = next(api.search(metadata_result_obj.sha256))
        print("Polyscore: {}".format(artifact_instance.polyscore))
        break

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}')