PolySwarmPolySwarmPolySwarmPolySwarm
Help

Building an EICAR-Submitting Ambassador

EICAR Test File

The EICAR test file is used by the Antivirus industry to test engines' ability to detect malware. The file is not malicious, but is flagged as such by most vendors.

Here we discuss implementing an Ambassador that submits the EICAR test file. Elsewhere, we discuss implementing a Microengine that detects the EICAR test file.


Submitting EICAR

polyswarm-client comes with an EICAR-submitting Ambassador (eicar.py) that hosts artifacts on IPFS, a public, distributed file sharing network.

eicar.py begins:

import base64
import logging
import random
import os

from concurrent.futures import CancelledError

from polyswarmartifact import ArtifactType

from polyswarmclient.abstractambassador import AbstractAmbassador

logger = logging.getLogger(__name__)

EICAR = base64.b64decode(
    b'WDVPIVAlQEFQWzRcUFpYNTQoUF4pN0NDKTd9JEVJQ0FSLVNUQU5EQVJELUFOVElWSVJVUy1URVNULUZJTEUhJEgrSCo=')
NOT_EICAR = 'this is not malicious'
ARTIFACTS = [('eicar', EICAR), ('not_eicar', NOT_EICAR)]

After some imports and logging configuration, the EICAR string and a string that is decidedly not EICAR are hard-coded. These strings are placed in an ARTIFACTS array.

Continuing:

BOUNTY_TEST_DURATION_BLOCKS = int(os.getenv('BOUNTY_TEST_DURATION_BLOCKS', 5))

eicar.py sets the default assertion duration window to 5 blocks. Block duration in wall time is decided by those hosting the community. In Swarm Technologies-hosted communities, 1 block is added to the chain approximately every second, so a 5 block window is approximately 5 seconds. This default can be overridden with an environment variable.

class Ambassador(AbstractAmbassador):
    """Ambassador which submits the EICAR test file"""

    def __init__(self, client, testing=0, chains=None, watchdog=0, submission_rate=30):
        """
        Initialize {{ cookiecutter.participant_name }}
        Args:
            client (`Client`): Client to use
            testing (int): How many test bounties to respond to
            chains (set[str]): Chain(s) to operate on
            watchdog: interval over which a watchdog thread should verify bounty placement on-chain (in number of blocks)
            submission_rate: if nonzero, produce a sleep in the main event loop to prevent the ambassador from overloading `polyswarmd` during testing
        """
        super().__init__(client, testing, chains, watchdog, submission_rate)

eicar.py's Ambassader is built on polyswarm-client's AbstractAmbassador. Among many other things, AbstractAmbassador establishes a connection to a hosted polyswarmd and manages this connection through an instance variable named client.

AbstractAmbassador declares a single method, generate_bounties, as abstract. All subclasses of AbstractAmbassador must define this method.

As you might suspect, eicar.py's implementation of this method is rather straightforward:

    async def generate_bounties(self, chain):
        """Submit either the EICAR test string or a benign sample
        Args:
            chain (str): Chain sample is being requested from
        """
        amount = await self.client.bounties.parameters[chain].get('bounty_amount_minimum')

        while True:
            try:
                filename, content = random.choice(ARTIFACTS)

                logger.info('Submitting %s', filename)
                ipfs_uri = await self.client.post_artifacts([(filename, content)])
                if not ipfs_uri:
                    logger.error('Error uploading artifact to IPFS, continuing')
                    continue

                await self.push_bounty(ArtifactType.FILE, amount, ipfs_uri, BOUNTY_TEST_DURATION_BLOCKS, chain)
            except CancelledError:
                logger.info('Cancel requested')
                break
            except Exception:
                logger.exception('Exception in bounty generation task, continuing')
                continue

The method does the following (modulo error checking):

  1. queries polyswarmd for the minimum initial bounty amount
  2. enters an infinite loop
  3. randomly chooses either the eicar or the not_eicar string as the artifact
  4. tells polyswarmd to host the artifact on IPFS
  5. instructs polyswarmd to post the bounty, specifying the initial bounty amount (the minimum allowed), the URI of the artifact, the assertion window duration and the chain*

*chain refers to which blockchain to post the bounty on: "homechain" or "sidechain". This argument should always be side; it will be removed in a future polyswarm-client release.

Notes

  1. polyswarm-client-derived ambassadors are multi-threaded by default handling events asynchronously. This infinite loop will be isolated to the thread responsible for posting bounties; the remainder of the ambassador will function normally.
  2. There is no explicit sleep in the loop. This is intentional; the thread responsible for generate_bounties effectively sleeps while blocking on bounty submission each time it calls self.client.post_artifacts (which in turn blocks on IPFS hosting) and self.client.push_bounty (which blocks on the announcement of the bounty in the marketplace by polyswarmd).

eicar.py is a trivial example that does not account for many real-world ambassador operating concerns.

Customize Your Ambassador

Here we'll implement a simple, minimum viable ambassador, re-creating the EICAR ambassador described above.

Implementing a minimum viable Ambassador is as simple as implementing your Ambassador's generate_bounties method. This method is found in ambassador_<participant_name_slug>/src/<author_org_slug>_<participant_name_slug>/__init__.py.

Customize __init__.py to include the EICAR and not-EICAR definitions we saw in the EICAR ambassador:

...
logger = logging.getLogger(__name__)

EICAR = base64.b64decode(
    b'WDVPIVAlQEFQWzRcUFpYNTQoUF4pN0NDKTd9JEVJQ0FSLVNUQU5EQVJELUFOVElWSVJVUy1URVNULUZJTEUhJEgrSCo=')
NOT_EICAR = 'this is not malicious'
ARTIFACTS = [('eicar', EICAR), ('not_eicar', NOT_EICAR)]

BOUNTY_TEST_DURATION_BLOCKS = int(os.getenv('BOUNTY_TEST_DURATION_BLOCKS', 5))
...

Then, customize the generate_bounty method to submit EICAR and not-EICAR:

    async def generate_bounties(self, chain):
        """Submit either the EICAR test string or a benign sample
        Args:
            chain (str): Chain sample is being requested from
        """
        amount = await self.client.bounties.parameters[chain].get('bounty_amount_minimum')

        while True:
            try:
                filename, content = random.choice(ARTIFACTS)

                logger.info('Submitting %s', filename)
                ipfs_uri = await self.client.post_artifacts([(filename, content)])
                if not ipfs_uri:
                    logger.error('Error uploading artifact to IPFS, continuing')
                    continue

                await self.push_bounty(ArtifactType.FILE, amount, ipfs_uri, BOUNTY_TEST_DURATION_BLOCKS, chain)
            except CancelledError:
                logger.info('Cancel requested')
                break
            except Exception:
                logger.exception('Exception in bounty generation task, continuing')
                continue

Once these changes are made, you now have an EICAR ambassador built on participant-template!

Test Your Participant

Once everything is in place, let's test our participant:

Unit Testing →

Next Steps

An Ambassador that submits a hardcoded string is an unrealistic Ambassador. In the next section, we create a slightly more realistic Ambassador - one that submits files from a directory on disk.

Continue to File-Submitting Ambassador →