Simples Script to deactivate a user and prune it's messages from a Rocket.Chat Server

rocketchat.user.deactivator.py

# LICENSE: MIT
# Author: [email protected]
# 
# Required permissions:
# - manage-users
# - view-other-user-channels
# - remove messages
#
# Description: This script will:
# - Get provided users from a Rocket.Chat Server
# - Deactivate those users
# - List all messages from that user in channels and direct
# - Delete all those messages

# $ python3 rocketchat-user-deactivator.py -h
# usage: rocketchat-user-deactivator.py [-h] [-dry] usernames

# Rocket.Chat User Pruner

# positional arguments:
#   usernames        Usernames to inspect, optionally comma separated

# optional arguments:
#   -h, --help       show this help message and exit
#   -dry, --dry-run  dry/test run. Only shows what will be done (default: False)


import argparse
import requests
import os

ROCKETCHAT_HOST = os.environ.get(
    "ROCKETCHAT_USER_DEACTIVATOR_HOST", "http://localhost:3000")
ROCKETCHAT_USERID = os.environ.get(
    "ROCKETCHAT_USER_DEACTIVATOR_USERID", "avmX38wqkezv55obK")
ROCKETCHAT_TOKEN = os.environ.get(
    "ROCKETCHAT_USER_DEACTIVATOR_TOKEN", "dtOVV3ZQDZJp4BCG1YOINt7jQyMTuFeJ-p_mtyXGbvt")

headers = {
    "X-Auth-Token": ROCKETCHAT_TOKEN,
    "X-User-Id": ROCKETCHAT_USERID,
}


parser = argparse.ArgumentParser(description="Rocket.Chat User Pruner",
                                 formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("usernames", help="Usernames to inspect, optionally comma separated")
parser.add_argument("-dry", "--dry-run",
                    help="dry/test run. Only shows what will be done", action="store_true")
parser.add_argument("-remove", "--remove",
                    help="Remove the user, register and deactivate, in order to remove the messages", action="store_true")                    
args = parser.parse_args()
config = vars(args)


def get_user(username):
    url_get_user = ROCKETCHAT_HOST + \
        '/api/v1/users.info?fields={"userRooms": 1}&username=' + username
    return requests.get(url_get_user, headers=headers)


def deactivate_user(user_id):
    url_update_user = ROCKETCHAT_HOST + "/api/v1/users.update"
    payload = {
        "userId": user_id,
        "data": {
            "active": False
        }
    }
    return requests.post(url_update_user, headers=headers, json=payload)


def get_user_messages(roomId, user):
    url_search_user_messages = ROCKETCHAT_HOST + "/api/v1/chat.search?roomId={0}&searchText=from:{1}".format(
        roomId,
        user
    )
    return requests.get(url_search_user_messages, headers=headers)


def delete_message(roomId, msgId):
    url_delete_messages = ROCKETCHAT_HOST + "/api/v1/chat.delete"
    payload = {
        "roomId": roomId,
        "msgId": msgId
    }
    return requests.post(url_delete_messages, json=payload, headers=headers)


if __name__ == "__main__":
    print("Config: " + str(config))

    for user in config["usernames"].split(","):
        response = get_user(user)
        if response.status_code == 200:
            print(
                "#" * 40 + " USERNAME {0} ({1})".format(user, response.json()["user"]["_id"]))
            user_id = response.json()["user"]["_id"]
            if response.json()['user']['active']:
                if config['dry_run']:
                    print("Dry run: deactivating user")
                else:
                    response_deact = deactivate_user(user_id)
                    if response_deact.status_code == 200:
                        print("Deactivated!")
                    else:
                        print("Failed to deactivate :(")
            else:
                print("Already deactivated")
            # MESSAGE REMOVAL
            if response.json()["user"].get("rooms"):
                rooms = response.json()["user"]["rooms"]
                rooms_name = [r["name"] for r in rooms]
                print("Found {0} in {1} rooms: {2}".format(
                    user, len(rooms), ",".join(rooms_name)))
                for room in rooms:
                    print("# Checking room {0} ({1}). Type:".format(
                        room["name"], room["rid"]), room["t"])
                    response = get_user_messages(room["rid"], user)
                    messages = response.json()
                    if messages["messages"]:
                        print("found {0} messages".format(
                            len(messages["messages"])
                        ))
                        for msg in messages["messages"]:
                            print("\n" + "-" * 10 +
                                " MESSAGE {0} ({1})".format(msg["_id"], msg["ts"]))
                            print("- LINK: {0}/channel/{1}?msg={2} \n- BODY: {3}".format(
                                ROCKETCHAT_HOST,
                                room["name"],
                                msg["_id"],
                                msg["msg"]
                            ))
                            if config['dry_run']:
                                print("- DELETED: False *(dry run)*")
                            else:
                                delete_request = delete_message(
                                    room["rid"], msg["_id"])
                                if delete_request.ok:
                                    print("- DELETED: True (got 200 from server)")
                                else:
                                    print(
                                        "- DELETED: False (got !200 from server): {0}".format(
                                            delete_request.json()
                                        )
                                    )
                    else:
                        print("no messages found")
            else:
                print("Could not find what rooms the user is. Check if you have view-other-user-channels permission")
        elif response.status_code == 400:
            print("#" * 40 + " USER {0} NOT FOUND!".format(user))
        else:
            print("Error: " + str(response.status_code) + " " + response.text)