Cryptominer Research

[mssalvatore]

Spike

After the plugin interface for payloads has been defined, it should be exercised by adding a new payload. A Cryptominer simulation would be valuable to users.

In order to provide the most valuable cryptominer simulation, we should understand what behaviors or properties of the malware are the most common triggers for detection tools.

Objective

Understand the most common cryptominer detection mechanisms. Determine what behaviors a cryptominer payload should exhibit.

Output

1. A list of features the cryptominor v1.0 plugin should have
2. A list of options the user will be able to modify at runtime
3. An explanation of why these features were included
4. An explanation of why certain features were omitted
5. Technical details/resources that helped make these decisions
6. A list of future work items.

Resources

https://github.com/Hestat/minerchk

https://github.com/jsiwek/zeek-cryptomining

[shreyamalviya]

Cryptojacking research notes

[shreyamalviya]

We can do something similar to this for consuming CPU and RAM.

[mssalvatore]

The SHA256 algorithm from python's library appears to be 1 order of Magnitude faster than the one from cryptography:

from timeit import timeit

print(
    timeit(
        "digest=hashes.Hash(hashes.SHA256());digest.update(nonce.to_bytes(3)); nonce +=1",
        setup="from cryptography.hazmat.primitives import hashes; nonce = 0",
        number=100000,
    )
)
print(
    timeit(
        "digest=hashlib.sha256();digest.update(nonce.to_bytes(3)); nonce +=1",
        setup="import hashlib; nonce = 0",
        number=100000,
    )
)

Output:

0.3059965530410409
0.018636079970747232

[shreyamalviya]

Features that could be included:

1. Excessive CPU usage
2. Excessive RAM usage
3. Cryptomining-specific API calls
4. Unexpected system calls
5. Excessive network traffic
6. Abnormal network traffic (packet sizes, interarrival times, etc)

We can't simulate in-browser cryptojacking through the Agent at this point.

---

Out of these, options 1 and 2 are simple to implement. Options 3 to 6 require in-depth research on the working of cryptominers. Options 5 and 6 will further need to be evaluated with respect to stealth.

---

For v1 of the plugin, I suggest we go with implementing options 1 and 2, where the user can configure the time for which the simulation will run, RAM usage in % (or MBs? we'll need a fallback in cases where that much RAM isn't available), and CPU usage in % (or number of processes? same concern as previous). The plugin can be extended to options 3 and 4 in future versions, and options 5 and 6, if possible.

[mssalvatore]

I think we should seriously investigate #3 for v1. Can you give more information about #4? What makes a system call "unexpected"?

[shreyamalviya]

Not sure. What I meant was system call events because of the presence of a cryptojacker would be unexpected.

[shreyamalviya]

Some things we could do:

1. Download cryptominers (lolMiner, XMRig) from GitHub (malware often does this)
2. Make calls to the Bitcoin API (Python interface)
3. Make requests to mining pools (list of mining pools here and here)
4. Start a process with an obvious name (like the ones this script checks for)

[shreyamalviya]

A good article with the breakdown of a cryptomining attack

[shreyamalviya]

We could also use the masquerade feature here. The configuration can have a checkbox for including the hashes of known cryptominers in the binaries.

[shreyamalviya]

Zeek (with the zeek-cryptomining script loaded) recognizes any request with "getblocktemplate" as the request method as cryptomining activity. We can add a configuration option to send out such a request to the Island server.

The script I used to test with Zeek:

(A modification of this)

import urllib.request
import urllib.error
import urllib.parse
import base64
import json
import random
import os

RPC_URL = os.environ.get("RPC_URL", "http://google.com/")
RPC_USER = os.environ.get("RPC_USER", "bitcoinrpc")
RPC_PASS = os.environ.get("RPC_PASS", "")

def rpc(method, params=None):
    rpc_id = random.getrandbits(32)
    data = json.dumps({"id": rpc_id, "method": method, "params": params}).encode()
    auth = base64.encodebytes((RPC_USER + ":" + RPC_PASS).encode()).decode().strip()

    request = urllib.request.Request(RPC_URL, data, {"Authorization": "Basic {:s}".format(auth)})

    with urllib.request.urlopen(request) as f:
        response = json.loads(f.read())

    if response['id'] != rpc_id:
        raise ValueError("Invalid response id: got {}, expected {:u}".format(response['id'], rpc_id))
    elif response['error'] is not None:
        raise ValueError("RPC error: {:s}".format(json.dumps(response['error'])))

    return response['result']


def rpc_getblocktemplate():
    try:
        return rpc("getblocktemplate", [{"rules": ["segwit"]}])
    except ValueError:
        return {}


rpc_getblocktemplate()