dpoll audit

dpoll_audit.py

import json
import re
from datetime import datetime, timedelta
from pprint import pprint

import requests
from lightsteem.client import Client

POLL_AUTHOR = "theycallmedan"
POLL_PERMLINK = "which-steem-project-should-i-delegate-10k-steempower-to-for-1-year"

c = Client(read_timeout=60, connect_timeout=30)

# Get the voter list from dPoll API
dpoll_api_resp = requests.get(
    f"https://dpoll.xyz/api/v1/questions/{POLL_AUTHOR}/?permlink={POLL_PERMLINK}").json()

choices = {}
voted_users = []
for choice in dpoll_api_resp["choices"]:
    voted_users += [u["username"] for u in choice["voted_users"]]
    choices[choice["text"].strip()] = [
        u["username"] for u in choice["voted_users"]]

replies = c.get_content_replies(POLL_AUTHOR, POLL_PERMLINK)

# Check for the get_content_replies to detect discrepancies
discrepancies = {}
for r in replies:
    try:
        meta = json.loads(r['json_metadata'])
    except Exception as e:
        continue
    for vote in meta.get("votes", []):
        choices[vote].remove(r["author"])

print("Possible discrepancies")
pprint(choices)
print("-" * 42)
stop_at = datetime.utcnow() - timedelta(days=6)
print("Checking account history for comment deletions")
for choice, voters in choices.items():
    for voter in voters:
        print(f"Checking @{voter}")
        delete_comment_found = False
        for row in c.account(voter).history(filter=["delete_comment"],
                                           stop_at=stop_at,
                                           only_operation_data=False):
            trx_id = row[1]["trx_id"]
            block_id = row[1]["block"]
            op = row[1]["op"][1]
            if re.match("[\w]{8}(-[\w]{4}){3}-[\w]{12}", op["permlink"]):
                delete_comment_found = True
                print(f"Found deleted comment by {op['author']}. "
                      f"(TRX id: {trx_id}, block id: {block_id}")
                break

        if not delete_comment_found:
            # if we can't find a comment deletion,
            # check for overwrites: E.G: partiko overwrites the json_metadata
            found_overwrite = False
            for row in c.account(voter).history(filter=["comment"],
                                               stop_at=stop_at,
                                               only_operation_data=False):
                trx_id = row[1]["trx_id"]
                block_id = row[1]["block"]
                op = row[1]["op"][1]
                if op["parent_permlink"] == POLL_PERMLINK and \
                        op["parent_author"] == POLL_AUTHOR:
                    print(op)
                    try:
                        meta = json.loads(op['json_metadata'])
                    except Exception as e:
                        continue

                    print(f"{op['author']}: {meta['app']}")
                    if not meta.get("app").startswith("dpoll"):
                        found_overwrite = True
                        print(f"Found overwrite by @{meta['app']}! "
                              f"(TRX id: {trx_id}, block id: {block_id}")
                        break

            if not found_overwrite:
                print(f"DPOLL BUG. Couldn't find any reference for @{voter}")