prod_rmq_agents/get-next-uid-gid.py

+#!/usr/bin/env python
+import json
+import time
+import rc_util
+from os import popen
+from rc_rmq import RCRMQ
+import rabbit_config as rcfg
+from subprocess import run
+import shlex
+
+task = "create_account"
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+args = rc_util.get_args()
+
+# Logger
+logger = rc_util.get_logger()
+
+
+# Account creation
+def create_account(msg):
+
+    logger.info(f"Account creation request received: {msg}")
+    username = msg["username"]
+    uid = msg["uid"]
+    email = msg["email"]
+    fullname = msg["fullname"]
+    msg["success"] = False
+
+    # Bright command to create user
+    if str(rcfg.bright_cm_version).split(".")[0] == "8":
+        cmd = "/cm/local/apps/cmd/bin/cmsh -c "
+        cmd += f'"user; add {username}; set userid {uid}; set email {email};'
+        cmd += f'set commonname \\"{fullname}\\"; '
+        cmd += 'commit;"'
+    else:
+        cmd = "/cm/local/apps/cmd/bin/cmsh -c "
+        cmd += f'"user; add {username}; set id {uid}; set email {email};'
+        cmd += f'set commonname \\"{fullname}\\"; '
+        cmd += 'commit;"'
+
+    if not args.dry_run:
+        run(shlex.split(cmd))
+        time.sleep(rcfg.Delay)
+    logger.info(f"Bright command to create user:{cmd}")
+
+
+# Define your callback function
+def resolve_uid_gid(ch, method, properties, body):
+
+    # Retrieve message
+    msg = json.loads(body)
+    logger.info("Received {}".format(msg))
+    username = msg["username"]
+    msg["success"] = False
+
+    # Determine next available UID
+    try:
+        user_exists_cmd = f"/usr/bin/getent passwd {username}"
+        user_exists = popen(user_exists_cmd).read().rstrip()
+
+        if user_exists:
+            logger.info("The user, {} already exists".format(username))
+            msg["uid"] = user_exists.split(":")[2]
+            msg["gid"] = user_exists.split(":")[3]
+
+        else:
+            cmd_uid = (
+                "/usr/bin/getent passwd | awk -F: 'BEGIN { maxuid=10000 }"
+                " ($3>10000) && ($3<20000) && ($3>maxuid) { maxuid=$3; } END {"
+                " print maxuid+1; }'"
+            )
+            msg["uid"] = popen(cmd_uid).read().rstrip()
+            logger.info(f"UID query: {cmd_uid}")
+
+            cmd_gid = (
+                "/usr/bin/getent group | awk -F: 'BEGIN { maxgid=10000 }"
+                " ($3>10000) && ($3<20000) && ($3>maxgid) { maxgid=$3; } END {"
+                " print maxgid+1; }'"
+            )
+            msg["gid"] = popen(cmd_gid).read().rstrip()
+            logger.info(f"GID query: {cmd_gid}")
+
+            create_account(msg)
+        msg["task"] = task
+        msg["success"] = True
+    except Exception:
+        msg["success"] = False
+        msg["errmsg"] = (
+            "Exception raised during account creation, check logs for stack"
+            " trace"
+        )
+        logger.error("", exc_info=True)
+
+    # Acknowledge message
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+    # Send confirm message
+    logger.debug("rc_rmq.publish_msg()")
+    rc_rmq.publish_msg(
+        {"routing_key": "confirm." + msg["queuename"], "msg": msg}
+    )
+    logger.info("confirmation sent")
+
+
+logger.info("Start listening to queue: {}".format(task))
+rc_rmq.start_consume(
+    {"queue": task, "routing_key": "request.*", "cb": resolve_uid_gid}
+)
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/git_commit.py

+#!/usr/bin/env python
+import os
+import sh
+import json
+import rc_util
+from rc_rmq import RCRMQ
+import rabbit_config as rmq_cfg
+import time
+
+task = "git_commit"
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+# Define some location
+repo_location = os.path.expanduser(rmq_cfg.rc_users_ldap_repo_loc)
+users_dir = repo_location + "/users"
+groups_dir = repo_location + "/groups"
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+if not args.dry_run:
+    git = sh.git.bake(
+        "--git-dir", repo_location + "/.git", "--work-tree", repo_location
+    )
+    ldapsearch = sh.Command("ldapsearch")
+else:
+    git = sh.echo.bake(
+        "--git-dir", repo_location + "/.git", "--work-tree", repo_location
+    )
+    ldapsearch = sh.echo.bake("ldapsearch")
+
+
+def git_commit(ch, method, properties, body):
+    msg = json.loads(body)
+    username = msg["username"]
+    msg["task"] = task
+    msg["success"] = False
+    branch_name = (
+        "issue-add-users-"
+        + username.lower()
+        + "-"
+        + time.strftime("%Y%m%d_%H%M%S")
+    )
+    user_ldif = users_dir + f"/{username}.ldif"
+    group_ldif = groups_dir + f"/{username}.ldif"
+
+    logger.info("Received: %s", msg)
+    logger.debug("branch_name: %s", branch_name)
+
+    try:
+
+        logger.debug("git checkout master")
+        git.checkout("master")
+        logger.debug("git pull")
+        git.pull()
+        branch_exists = git.branch("--list", branch_name)
+        if not branch_exists:
+            logger.debug("git checkout -b %s", branch_name)
+            git.checkout("-b", branch_name)
+            logger.debug("open(%s, 'w'), open(%s, 'w')", user_ldif, group_ldif)
+            with open(user_ldif, "w") as ldif_u, open(
+                group_ldif, "w"
+            ) as ldif_g:
+                logger.debug(
+                    "ldapsearch -LLL -x -H ldaps://ldapserver -b 'dc=cm,dc=clu"
+                    f"ster' uid={username} > {user_ldif}"
+                )
+                ldapsearch(
+                    "-LLL",
+                    "-x",
+                    "-H",
+                    "ldaps://ldapserver",
+                    "-b",
+                    "dc=cm,dc=cluster",
+                    f"uid={username}",
+                    _out=ldif_u,
+                )
+                logger.debug(
+                    "ldapsearch -LLL -x -H ldapserver -b 'ou=Group,dc=cm,dc=cl"
+                    f"uster' cn={username} > {group_ldif}"
+                )
+                ldapsearch(
+                    "-LLL",
+                    "-x",
+                    "-H",
+                    "ldaps://ldapserver",
+                    "-b",
+                    "ou=Group,dc=cm,dc=cluster",
+                    f"cn={username}",
+                    _out=ldif_g,
+                )
+            logger.info("user ldif files generated.")
+
+            logger.debug("git add %s", user_ldif)
+            git.add(user_ldif)
+            logger.debug("git add %s", group_ldif)
+            git.add(group_ldif)
+            logger.debug("git commit -m 'Added new cheaha user: %s'", username)
+            git.commit(m="Added new cheaha user: " + username)
+            logger.debug("git checkout master")
+            git.checkout("master")
+
+            logger.debug("git merge %s --no-ff --no-edit", branch_name)
+            git.merge(branch_name, "--no-ff", "--no-edit")
+            logger.debug("git push origin master")
+            git.push("origin", "master")
+            # merge with gitlab api
+
+            logger.info("Added ldif files and committed to git repo")
+
+        msg["success"] = True
+    except Exception:
+        logger.error("", exc_info=True)
+
+    # Send confirm message
+    logger.debug("rc_rmq.publish_msge()")
+    rc_rmq.publish_msg(
+        {"routing_key": "confirm." + msg["queuename"], "msg": msg}
+    )
+    logger.info("confirmation sent")
+
+    # Acknowledge message
+    logger.debug("ch.basic_ack()")
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+logger.info("Start listening to queue: %s", task)
+rc_rmq.start_consume(
+    {"queue": task, "routing_key": "verify.*", "cb": git_commit}
+)
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/group_member.py

+#!/usr/bin/env python
+import dataset
+import json
+import pika
+import shlex
+import rc_util
+from datetime import datetime
+from subprocess import Popen, PIPE
+from rc_rmq import RCRMQ
+import rabbit_config as rcfg
+
+task = "group_member"
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+# Initialize db
+db = dataset.connect(f"sqlite:///{rcfg.db_path}/user_reg.db")
+table = db["groups"]
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": rcfg.Exchange, "exchange_type": "topic"})
+
+
+def insert_db(operation, groupname, msg):
+    if operation == "remove":
+        op = 0
+    elif operation == "add":
+        op = 1
+
+    # SQL insert
+    table.insert(
+        {
+            "user": msg["username"],
+            "group": groupname,
+            "operation": op,
+            "date": datetime.now(),
+            "host": msg["host"],
+            "updated_by": msg["updated_by"],
+            "interface": msg.get("interface", ""),
+        }
+    )
+
+
+def group_member(ch, method, properties, body):
+    """
+    Properties:
+      correlation_id (str): The UUID for the request.
+      reply_to       (str): The RabbitMQ queue name for reply to send to.
+
+    Message(body):
+      username    (str): The user to be added/removed from groups.
+      groups     (dict): A dictionary with `add` or `remove` key.
+        add      (list): A list of groups to be added for the user.
+        remove   (list): A list of groups to be removed for the user.
+      updated_by  (str): The user who request the change.
+      host        (str): Hostname where the request comes from.
+      interface   (str): whether it's from CLI or WebUI.
+
+    Returns:
+      status (bool): Whether or not the operation executed successfully.
+      errmsg  (str): Detailed error message if operation failed.
+      task    (str): The task name of the agent who handle the message.
+    """
+    msg = json.loads(body)
+    username = msg["username"]
+    msg["task"] = task
+
+    try:
+        if "remove" in msg["groups"]:
+            for each_group in msg["groups"]["remove"]:
+                logger.debug(
+                    f"Removing user {username} from group {each_group}"
+                )
+                if str(rcfg.bright_cm_version).split(".")[0] == "8":
+                    grp_remove_user_cmd = (
+                        '/cm/local/apps/cmd/bin/cmsh -n -c "group; removefrom'
+                        f' {each_group} groupmembers {username}; commit;"'
+                    )
+                else:
+                    grp_remove_user_cmd = (
+                        '/cm/local/apps/cmd/bin/cmsh -n -c "group; removefrom'
+                        f' {each_group} members {username}; commit;"'
+                    )
+
+                logger.info(f"Running command: {grp_remove_user_cmd}")
+                proc = Popen(
+                    shlex.split(grp_remove_user_cmd), stdout=PIPE, stderr=PIPE
+                )
+                out, err = proc.communicate()
+                logger.debug(f"Result: {err}")
+                logger.info(
+                    f"User {username} is removed from {each_group} group"
+                )
+                insert_db("remove", each_group, msg)
+
+        if "add" in msg["groups"]:
+            for each_group in msg["groups"]["add"]:
+                logger.debug(f"Adding user {username} to group {each_group}")
+                if str(rcfg.bright_cm_version).split(".")[0] == "8":
+                    grp_add_user_cmd = (
+                        '/cm/local/apps/cmd/bin/cmsh -n -c "group; append'
+                        f' {each_group} groupmembers {username}; commit;"'
+                    )
+                else:
+                    grp_add_user_cmd = (
+                        '/cm/local/apps/cmd/bin/cmsh -n -c "group; append'
+                        f' {each_group} members {username}; commit;"'
+                    )
+
+                logger.info(f"Running command: {grp_add_user_cmd}")
+                proc = Popen(
+                    shlex.split(grp_add_user_cmd), stdout=PIPE, stderr=PIPE
+                )
+                out, err = proc.communicate()
+                logger.debug(f"Result: {err}")
+                logger.info(f"User {username} is added to {each_group} group")
+                insert_db("add", each_group, msg)
+
+        msg["success"] = True
+
+    except Exception:
+        msg["success"] = False
+        msg["errmsg"] = (
+            "Exception raised, while adding user to group {groupname}, check"
+            " the logs for stack trace"
+        )
+        logger.error("", exc_info=True)
+
+    corr_id = properties.correlation_id
+    reply_to = properties.reply_to
+
+    logger.debug(f"corr_id: {corr_id} \n reply_to: {reply_to}")
+    # send response to the callback queue
+    if reply_to:
+        props = pika.BasicProperties(correlation_id=corr_id)
+        logger.debug("Sending confirmation back to reply_to")
+        rc_rmq.publish_msg(
+            {"routing_key": reply_to, "props": props, "msg": msg}
+        )
+    else:
+        print("Error: no reply_to")
+
+    logger.debug(f"User {username} confirmation sent from {task}")
+
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+logger.info(f"Start listening to queue: {task}")
+rc_rmq.bind_queue(queue=task, routing_key="group_member.*", durable=True)
+
+rc_rmq.start_consume({"queue": task, "cb": group_member})
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/new_jobs.py

+#!/usr/bin/env python
+import os
+import json
+import pika
+import rc_util
+from os import popen
+from pathlib import Path
+from rc_rmq import RCRMQ
+import rabbit_config as rcfg
+
+task = "new_jobs"
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": rcfg.Exchange, "exchange_type": "topic"})
+
+
+def new_jobs(ch, method, properties, body):
+    msg = json.loads(body)
+    username = msg["username"]
+    action = msg["action"]
+    msg["task"] = task
+    queuename = msg["queuename"]
+
+    try:
+        block_new_jobs_cmd = f"/cm/shared/apps/slurm/19.05.5/bin/sacctmgr --immediate update user {username} set maxjobs=0"
+        unblock_new_jobs_cmd = f"/cm/shared/apps/slurm/19.05.5/bin/sacctmgr --immediate update user {username} set maxjobs=-1"
+
+        if action == 'lock':
+            block_new_jobs = popen(block_new_jobs_cmd).read().rstrip()
+        elif action == 'unlock':
+            unblock_new_jobs = popen(unblock_new_jobs_cmd).read().rstrip()
+
+        msg["success"] = True
+        logger.info(f"Succeeded in blocking {username}'s jobs getting to run state")
+
+    except Exception:
+        msg["success"] = False
+        msg["errmsg"] = "Exception raised while setting maxjobs that can enter run state, check the logs for stack trace"
+        logger.error("", exc_info=True)
+
+
+    rc_rmq.publish_msg(
+        {"routing_key": f'acctmgr.done.{queuename}', 
+         "msg": msg}
+    )
+
+    logger.debug(f"User {username} confirmation sent for {action}ing {task}")
+
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+logger.info(f"Start listening to queue: {task}")
+rc_rmq.bind_queue(queue=task, routing_key='lock.*', durable=True)
+rc_rmq.bind_queue(queue=task, routing_key='unlock.*', durable=True)
+rc_rmq.bind_queue(queue=task, routing_key='newjobs.*', durable=True)
+rc_rmq.start_consume(
+    {"queue": task, "cb": new_jobs}
+)
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/notify_user.py

+#!/usr/bin/env python
+import json
+import rc_util
+import smtplib
+import dataset
+from rc_rmq import RCRMQ
+from jinja2 import Template
+from datetime import datetime
+import rabbit_config as rcfg
+import mail_config as mail_cfg
+
+task = "notify_user"
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+db = dataset.connect(f"sqlite:///{rcfg.db_path}/user_reg.db")
+table = db["users"]
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+
+# Email instruction to user
+def notify_user(ch, method, properties, body):
+    msg = json.loads(body)
+    username = msg["username"]
+    user_email = msg["email"]
+    msg["task"] = task
+    msg["success"] = False
+    errmsg = ""
+
+    try:
+
+        # Search username in database
+        record = table.find_one(username=username)
+
+        if record["sent"] is not None:
+            errmsg = "Updating database counter"
+            # Update counter
+            count = record["count"]
+            if args.dry_run:
+                logger.info("Update counter in database")
+
+            else:
+                table.update(
+                    {"username": username, "count": count + 1},
+                    ["username"],
+                )
+
+            logger.debug(f"User {username} counter updated to {count + 1}")
+
+        else:
+            # Send email to user
+            receivers = [user_email, rcfg.Admin_email]
+            message = Template(mail_cfg.Whole_mail).render(
+                username=username, to=user_email
+            )
+
+            if args.dry_run:
+                logger.info(f"smtp = smtplib.SMTP({rcfg.Mail_server})")
+                logger.info(
+                    f"smtp.sendmail({rcfg.Sender}, {receivers}, message)"
+                )
+                logger.info(
+                    f"table.update({{'username': {username}, 'count': 1,"
+                    " 'sent_at': datetime.now()}}, ['username'])"
+                )
+
+            else:
+                errmsg = "Sending email to user"
+                smtp = smtplib.SMTP(rcfg.Mail_server)
+                smtp.sendmail(rcfg.Sender_notification, receivers, message)
+
+                logger.debug(f"Email sent to: {user_email}")
+
+                errmsg = "Updating database email sent time"
+                table.update(
+                    {
+                        "username": username,
+                        "count": 1,
+                        "sent": datetime.now(),
+                    },
+                    ["username"],
+                )
+
+                logger.debug(f"User {username} inserted into database")
+
+        msg["success"] = True
+    except Exception:
+        logger.error("", exc_info=True)
+        msg["errmsg"] = errmsg if errmsg else "Unexpected error"
+
+    # Send confirm message
+    rc_rmq.publish_msg(
+        {"routing_key": "confirm." + msg["queuename"], "msg": msg}
+    )
+
+    logger.debug(f"User {username} confirmation sent")
+
+    # Acknowledge the message
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+if __name__ == "__main__":
+    logger.info(f"Start listening to queue: {task}")
+    rc_rmq.start_consume(
+        {"queue": task, "routing_key": "notify.*", "cb": notify_user}
+    )
+
+    logger.info("Disconnected")
+    rc_rmq.disconnect()

prod_rmq_agents/ssh_access.py

+#!/usr/bin/env python
+import os
+import json
+import pika
+import uuid
+import rc_util
+from subprocess import Popen,PIPE
+from pathlib import Path
+from rc_rmq import RCRMQ
+import rabbit_config as rcfg
+
+task = "ssh_access"
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": rcfg.Exchange, "exchange_type": "topic"})
+
+
+def ssh_access(ch, method, properties, body):
+    msg = json.loads(body)
+    routing_key = method.routing_key
+    username = msg["username"]
+    action = msg["action"]
+    msg["task"] = task
+    queuename = msg["queuename"]
+    state = msg["state"]
+
+    global corr_id
+    try:
+    # check if it's a response from group_member_agent
+        if routing_key == task:
+            logger.debug(f"corr_id sent by group_member agent: {properties.correlation_id}")
+            if corr_id == properties.correlation_id:
+                logger.debug(f'group_member agent confirmation msg["success"]: {msg["success"]}')
+                # forward confirmation response to acct_mgmt_workflow agent
+                rc_rmq.publish_msg(
+                   {
+                    "routing_key": f'acctmgr.done.{queuename}',
+                    "msg": msg
+                   }
+                )
+                logger.debug(f'User {username} confirmation sent for {action}ing {task}')
+                
+        else:
+            corr_id = str(uuid.uuid4())
+            logger.debug(f'corr_id generated: {corr_id}')
+            msg["groups"] = {}
+
+            proc = Popen(['/usr/bin/groups', username], stdout=PIPE, stderr=PIPE)
+            out,err = proc.communicate()
+
+            user_groups = out.decode().strip().split(":")[1].split()
+            state_groups = rcfg.state_groups
+            """
+              Filter the lock group a user is in and assign to spl
+              lambda function returns common elements between two lists. For all
+              the true values by returned lambda function for common elements
+              corresponding values are included as a list by filter function.
+            """
+            user_state_groups = list(filter(lambda x:x in list(rcfg.state_groups.values()),user_groups))
+
+            # Depending on state add user to the group corresponding to state.
+            # Remove user from user_state_groups they are already part of.
+            # eg: {"groups": { "add":[a,b,c], "remove":[d,e,f] }
+            if state == 'certification':
+                msg["groups"]["add"] = [state_groups[state]]
+                msg["groups"]["remove"] = user_state_groups
+
+            elif state == 'hold':
+                msg["groups"]["add"] = [state_groups[state]]
+                msg["groups"]["remove"] = user_state_groups
+
+            elif state == 'pre_certification':
+                msg["groups"]["add"] = [state_groups[state]]
+                msg["groups"]["remove"] = user_state_groups
+
+            elif state == 'ok':
+                msg["groups"]["remove"] = user_state_groups
+ 
+            # send a message to group_member.py agent
+            logger.debug(f"sending msg to group agent: {msg}")
+            rc_rmq.publish_msg(
+                {
+                   "routing_key": f'group_member.{queuename}',
+                   "props": pika.BasicProperties(
+                                correlation_id = corr_id,
+                                reply_to = task,
+                                ),
+                   "msg": msg
+                }
+            )
+            logger.info(f"Request sent to add/remove user {username} to/from spl groups")
+ 
+    except Exception:
+        msg["success"] = False
+        msg["errmsg"] = "Exception raised in ssh_access agent, check the logs for stack trace"
+        logger.error("", exc_info=True)
+
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+logger.info(f"Start listening to queue: {task}")
+rc_rmq.bind_queue(queue=task, routing_key='lock.*', durable=True)
+rc_rmq.bind_queue(queue=task, routing_key='unlock.*', durable=True)
+rc_rmq.bind_queue(queue=task, routing_key='ssh.*', durable=True)
+rc_rmq.bind_queue(queue=task, routing_key=task, durable=True)
+
+rc_rmq.start_consume(
+    {"queue": task, "cb": ssh_access}
+)
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/subscribe_mail_lists.py

+#!/usr/bin/env python
+import json
+import smtplib
+import rc_util
+from email.message import EmailMessage
+from rc_rmq import RCRMQ
+import rabbit_config as rcfg
+
+task = "subscribe_mail_list"
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+# Parse arguments
+args = rc_util.get_args()
+
+# Logger
+logger = rc_util.get_logger()  # Define your callback function
+
+
+def mail_list_subscription(ch, method, properties, body):
+
+    # Retrieve message
+    msg = json.loads(body)
+    logger.info("Received msg {}".format(msg))
+    username = msg["username"]
+    fullname = msg["fullname"]
+    email = msg["email"]
+
+    mail_list_admin = rcfg.Sender_subscription
+    mail_list = rcfg.Mail_list
+    mail_list_bcc = rcfg.Mail_list_bcc
+    server = rcfg.Mail_server
+
+    listserv_cmd = (
+        f"QUIET ADD hpc-announce {email} {fullname}\n"
+        f"QUIET ADD hpc-users {email} {fullname}"
+    )
+
+    logger.info("Adding user{} to mail list".format(username))
+    msg["success"] = False
+    try:
+        # Create a text/plain message
+        email_msg = EmailMessage()
+
+        email_msg["From"] = mail_list_admin
+        email_msg["To"] = mail_list
+        email_msg["Subject"] = ""
+        email_msg["Bcc"] = mail_list_bcc
+
+        # Create an smtp object and send email
+        s = smtplib.SMTP(server)
+
+        email_msg.set_content(listserv_cmd)
+        if not args.dry_run:
+            s.send_message(email_msg)
+        logger.info(
+            f"This email will add user {username} to listserv \n{email_msg}"
+        )
+
+        s.quit()
+        msg["task"] = task
+        msg["success"] = True
+    except Exception:
+        logger.error("", exc_info=True)
+
+    # Acknowledge message
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+    # send confirm message
+    logger.debug("rc_rmq.publish_msg()")
+    rc_rmq.publish_msg(
+        {"routing_key": "confirm." + msg["queuename"], "msg": msg}
+    )
+    logger.info("confirmation sent")
+
+
+logger.info("Start listening to queue: {}".format(task))
+rc_rmq.start_consume(
+    {
+        "queue": task,  # Define your Queue name
+        "routing_key": "verify.*",  # Define your routing key
+        "cb": mail_list_subscription,  # Pass callback function you just define
+    }
+)
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/task_manager.py

+#!/usr/bin/env python
+import copy
+import json
+import signal
+import dataset
+import rc_util
+import smtplib
+from rc_rmq import RCRMQ
+from jinja2 import Template
+from datetime import datetime
+import mail_config as mail_cfg
+import rabbit_config as rcfg
+
+task = "task_manager"
+timeout = 30
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+db = dataset.connect(f"sqlite:///{rcfg.db_path}/user_reg.db")
+table = db["users"]
+
+record = {
+    "uid": -1,
+    "gid": -1,
+    "email": "",
+    "reason": "",
+    "fullname": "",
+    "last_update": datetime.now(),
+    "errmsg": None,
+    "waiting": set(),
+    "request": {"create_account": None},
+    "verify": {
+        "git_commit": None,
+        "dir_verify": None,
+        "subscribe_mail_list": None,
+    },
+    "notify": {"notify_user": None},
+    "reported": False,
+}
+
+# Currently tracking users
+tracking = {}
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+
+def notify_admin(username, user_record):
+    receivers = [rcfg.Admin_email]
+
+    result = (
+        "SUCCESS"
+        if user_record["request"]["create_account"]
+        and user_record["verify"]["git_commit"]
+        and user_record["verify"]["dir_verify"]
+        and user_record["verify"]["subscribe_mail_list"]
+        and user_record["notify"]["notify_user"]
+        else "FAILED"
+    )
+
+    message = Template(mail_cfg.UserReportHead).render(
+        username=username, fullname=user_record["fullname"], result=result
+    )
+    if user_record["reported"]:
+        message += " (Duplicate)"
+    message += f""" \n
+    User Creation Report for user {username}
+    uid: {user_record["uid"]}, gid: {user_record["gid"]}
+    Tasks:
+    'create_account':      {user_record["request"]["create_account"]}
+    'git_commit':          {user_record["verify"]["git_commit"]}
+    'dir_verify':          {user_record["verify"]["dir_verify"]}
+    'subscribe_mail_list': {user_record["verify"]["subscribe_mail_list"]}
+    'notify_user':         {user_record["notify"]["notify_user"]}
+    """
+    if user_record["errmsg"]:
+        message += """
+
+        Error(s):
+        """
+        for msg in user_record["errmsg"]:
+            message += msg + "\n"
+
+    if args.dry_run:
+        logger.info(f"smtp = smtplib.SMTP({rcfg.Mail_server})")
+        logger.info(
+            f"smtp.sendmail({rcfg.Sender}, {rcfg.Admin_email}, message)"
+        )
+        logger.info(message)
+
+    else:
+        smtp = smtplib.SMTP(rcfg.Mail_server)
+        smtp.sendmail(rcfg.Sender, receivers, message)
+
+        logger.debug(f"User report sent to: {rcfg.Admin_email}")
+
+
+def insert_db(username, msg):
+    # Search username in db
+    record = table.find_one(username=username)
+
+    if not record:
+        # SQL insert
+        table.insert(
+            {
+                "username": username,
+                "uid": msg.get("uid", -1),
+                "gid": msg.get("gid", -1),
+                "email": msg.get("email", ""),
+                "reason": msg.get("reason", ""),
+                "fullname": msg.get("fullname", ""),
+                "create_account": None,
+                "git_commit": None,
+                "dir_verify": None,
+                "subscribe_mail_list": None,
+                "notify_user": None,
+                "sent": None,
+                "reported": False,
+                "last_update": datetime.now(),
+                "queuename": msg.get("queuename", ""),
+            }
+        )
+
+
+def update_db(username, data):
+    obj = {"username": username, **data}
+    table.update(obj, ["username"])
+
+
+def task_manager(ch, method, properties, body):
+    msg = json.loads(body)
+    queuename = method.routing_key.split(".")[1]
+    username = msg["username"]
+    task_name = msg["task"]
+    success = msg["success"]
+    send = completed = terminated = False
+    routing_key = ""
+
+    if username in tracking:
+        current = tracking[username]
+
+    else:
+        user_db = table.find_one(username=username)
+
+        current = tracking[username] = copy.deepcopy(record)
+        current["errmsg"] = []
+        current["queuename"] = (
+            user_db["queuename"] if user_db else msg["queuename"]
+        )
+        current["uid"] = user_db["uid"] if user_db else msg["uid"]
+        current["gid"] = user_db["gid"] if user_db else msg["gid"]
+        current["email"] = user_db["email"] if user_db else msg["email"]
+        current["reason"] = user_db["reason"] if user_db else msg["reason"]
+        current["fullname"] = (
+            user_db["fullname"] if user_db else msg["fullname"]
+        )
+
+        if user_db:
+            # Restore task status
+            current["request"]["create_account"] = user_db["create_account"]
+            current["verify"]["git_commit"] = user_db["git_commit"]
+            current["verify"]["dir_verify"] = user_db["dir_verify"]
+            current["verify"]["subscribe_mail_list"] = user_db[
+                "subscribe_mail_list"
+            ]
+            current["notify"]["notify_user"] = user_db["notify_user"]
+
+            current["reported"] = user_db["reported"]
+
+            for t in ["git_commit", "dir_verify", "subscribe_mail_list"]:
+                if user_db[t] is None:
+                    current["waiting"].add(t)
+
+            if not current["waiting"] and user_db["notify_user"] is None:
+                current["waiting"].add("notify_user")
+
+            logger.debug(f"Loaded user {username} from DB")
+
+        else:
+            insert_db(username, msg)
+
+            logger.debug(f"Tracking user {username}")
+
+    current["last_update"] = datetime.now()
+
+    # Update Database
+    update_db(
+        username,
+        {task_name: success, "last_update": current["last_update"]},
+    )
+
+    # Save error message if the task was failed
+    if not success:
+        errmsg = msg.get("errmsg", "")
+        if errmsg:
+            current["errmsg"].append(f"{task_name}: {errmsg}")
+
+    # Define message that's going to be published
+    message = {
+        "username": username,
+        "queuename": queuename,
+        "uid": current["uid"],
+        "gid": current["gid"],
+        "email": current["email"],
+        "reason": current["reason"],
+        "fullname": current["fullname"],
+    }
+
+    try:
+        if task_name in current["request"]:
+            current["request"][task_name] = success
+            routing_key = "verify." + queuename
+
+            # Terminate the process if failed
+            if not success:
+                terminated = True
+                routing_key = "complete." + queuename
+                message["success"] = False
+                message["errmsg"] = current["errmsg"]
+
+            send = True
+            current["waiting"] = {
+                "git_commit",
+                "dir_verify",
+                "subscribe_mail_list",
+            }
+            logger.debug(f"Request level {task_name}? {success}")
+
+        elif task_name in current["verify"]:
+            current["verify"][task_name] = success
+            current["waiting"].discard(task_name)
+            routing_key = "notify." + queuename
+
+            if not current["waiting"]:
+                send = True
+                current["waiting"] = {"notify_user"}
+
+            # Terminate if dir_verify failed and all agents has responsed
+            if send and not current["verify"]["dir_verify"]:
+                terminated = True
+                routing_key = "complete." + queuename
+                message["success"] = False
+                message["errmsg"] = current["errmsg"]
+
+            logger.debug(f"Verify level {task_name}? {success}")
+
+        elif task_name in current["notify"]:
+            current["notify"][task_name] = success
+            current["waiting"].discard(task_name)
+            routing_key = "complete." + queuename
+            message["success"] = success
+            message["errmsg"] = current["errmsg"]
+
+            send = True
+
+            # The whole creation process has completed
+            completed = True
+
+            logger.debug(f"Notify level {task_name}? {success}")
+
+    except Exception:
+        logger.error("", exc_info=True)
+
+    if send:
+        # Send trigger message
+        rc_rmq.publish_msg({"routing_key": routing_key, "msg": message})
+
+        if task_name == "create_account" and success and rcfg.default_groups:
+            rc_rmq.publish_msg(
+                {
+                    "routing_key": "group_member.add",
+                    "msg": {
+                        "groups": {"add": rcfg.default_groups},
+                        "username": username,
+                        "host": msg.get("host"),
+                        "updated_by": msg.get("updated_by"),
+                        "interface": msg.get("interface"),
+                    },
+                }
+            )
+        logger.debug(f"Trigger message '{routing_key}' sent")
+
+        logger.debug("Previous level messages acknowledged")
+
+    # Send report to admin
+    if completed or terminated:
+        notify_admin(username, current)
+
+        update_db(username, {"reported": True})
+
+        rc_util.update_state(
+            username, "ok", msg.get("updated_by"), msg.get("host")
+        )
+
+        tracking.pop(username)
+
+        logger.debug("Admin report sent")
+
+    # Acknowledge message
+    ch.basic_ack(method.delivery_tag)
+
+
+def timeout_handler(signum, frame):
+    current_time = datetime.now()
+    for user in tuple(tracking):
+        delta = current_time - tracking[user]["last_update"]
+
+        if delta.seconds > timeout:
+            rc_rmq.publish_msg(
+                {
+                    "routing_key": "complete." + user,
+                    "msg": {
+                        "username": user,
+                        "success": False,
+                        "errmsg": [
+                            "Timeout on "
+                            + ", ".join(tracking[user]["waiting"])
+                        ],
+                    },
+                }
+            )
+
+            notify_admin(user, tracking[user])
+
+            update_db(user, {"reported": True})
+
+            tracking.pop(user)
+
+
+# Set initial timeout timer
+signal.signal(signal.SIGALRM, timeout_handler)
+signal.setitimer(signal.ITIMER_REAL, timeout, timeout)
+
+logger.info(f"Start listening to queue: {task}")
+rc_rmq.start_consume(
+    {"queue": task, "routing_key": "confirm.*", "cb": task_manager}
+)
+
+logger.info("Disconnected")
+rc_rmq.disconnect()

prod_rmq_agents/user_reg_event_logger.py

+#!/usr/bin/env python
+import json
+from rc_rmq import RCRMQ
+
+task = "user_reg_event_log"
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+
+# Define your callback function
+def log_user_reg_events(ch, method, properties, body):
+    # Retrieve message
+    msg = json.loads(body)
+
+    # Retrieve routing key
+    routing_key = method.routing_key
+    print(f"Got a message with routing key: {routing_key}")
+    print(msg)
+
+    # Acknowledge message
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+print("Start listening to queue: {}".format(task))
+rc_rmq.start_consume(
+    {
+        "queue": task,  # Define your Queue name
+        "routing_key": "#",  # Define your routing key
+        "cb": log_user_reg_events,  # Pass in callback function you just define
+    }
+)

prod_rmq_agents/user_reg_logger.py

+#!/usr/bin/env python
+import json
+import dataset
+import rc_util
+from rc_rmq import RCRMQ
+from datetime import datetime
+import rabbit_config as rcfg
+
+# Define queue name
+task = "reg_logger"
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+
+# Parse arguments
+args = rc_util.get_args()
+
+# Logger
+logger = rc_util.get_logger()
+
+# Open registry table in DB
+db = dataset.connect(f"sqlite:///{rcfg.db_path}/reg_logger.db")
+account_req_table = db["registry"]
+
+
+# Define registration logger callback
+def log_registration(ch, method, properties, body):
+
+    account_req = json.loads(body)
+    account_req["req_time"] = datetime.now()
+    account_req_table.insert(account_req)
+    logger.info("logged account request for %s", account_req["username"])
+
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+logger.info("Start listening to queue: {}".format(task))
+
+# Start consuming messages from queue with callback function
+rc_rmq.start_consume(
+    {"queue": task, "routing_key": "request.*", "cb": log_registration}
+)

prod_rmq_agents/user_state.py

+#!/usr/bin/env python
+import json
+import rc_util
+import dataset
+import pika
+from rc_rmq import RCRMQ
+from datetime import datetime
+import rabbit_config as rcfg
+
+task = "user_state"
+
+args = rc_util.get_args()
+logger = rc_util.get_logger(args)
+
+db = dataset.connect(f"sqlite:///{rcfg.db_path}/user_reg.db")
+table = db["user_state"]
+
+# Instantiate rabbitmq object
+rc_rmq = RCRMQ({"exchange": rcfg.Exchange, "exchange_type": "topic"})
+
+
+def user_state(ch, method, properties, body):
+    msg = json.loads(body)
+    username = msg["username"]
+    updated_by = msg.get("updated_by")
+    host = msg.get("host")
+    op = msg["op"]
+    msg["success"] = False
+    errmsg = ""
+
+    corr_id = properties.correlation_id
+    reply_to = properties.reply_to
+
+    try:
+
+        if op == "get":
+            errmsg = "Getting latest state of {username}"
+            record = table.find_one(username=username, order_by="-date")
+
+            if record:
+                msg["state"] = record["state"]
+                logger.debug(
+                    f'The latest state of {username} is {msg["state"]}'
+                )
+            else:
+                msg["state"] = "no-account"
+
+        elif op == "post":
+            state = msg["state"]
+            errmsg = "Updating state of {username} to {state}"
+            table.insert(
+                {
+                    "username": username,
+                    "state": state,
+                    "date": datetime.now(),
+                    "updated_by": updated_by,
+                    "host": host,
+                }
+            )
+            logger.debug(f"User {username} state updates to {state}")
+
+        msg["success"] = True
+    except Exception:
+        logger.error("", exc_info=True)
+        msg["errmsg"] = errmsg if errmsg else "Unexpected error"
+
+    # Send response
+    if reply_to:
+        props = pika.BasicProperties(correlation_id=corr_id)
+        rc_rmq.publish_msg(
+            {"routing_key": reply_to, "msg": msg, "props": props}
+        )
+
+    # Acknowledge the message
+    ch.basic_ack(delivery_tag=method.delivery_tag)
+
+
+if __name__ == "__main__":
+    logger.info(f"Start listening to queue: {task}")
+    rc_rmq.start_consume({"queue": task, "cb": user_state})
+
+    logger.info("Disconnected")
+    rc_rmq.disconnect()

pyproject.toml

+[tool.black]
+line-length = 79
+target-version = ['py36']
+preview = true
+[tool.pylint.main]
+disable = ["invalid-name", "import-error", "unused-argument", "broad-except"]
+ignore = ["config.py", "tests.py"]
+[tool.pylint.format]
+max-line-length = 79

rabbit_config.py.example

@@ -4,3 +4,55 @@ Password = 'CHANGE_IT_TO_YOUR_OWN_PASSWORD'
 VHost = '/'
 Server = 'ohpc'
 Port = 5672
+
+Valid_state = ["ok", "blocked", "certification"]
+
+# Default function timeout
+Function_timeout = 30
+
+# time delay to let account creation finish
+# to avoid concurrency with downstream agents
+Delay = 5
+
+# dir_verify agent config
+User_dirs = ['/home', '/data/user', '/data/scratch']
+
+# git_commit agent config
+rc_users_ldap_repo_loc = "~/git/rc-users"
+db_path = ".agent_db"
+
+# Config related to email
+Mail_server = 'localhost'
+Admin_email = 'root@localhost'
+Sender = 'ROOT@LOCALHOST'
+Sender_notification = 'NOTIFY@LOCALHOST'
+Sender_subscription = 'SUBSCRIBE_EMAIL@LOCALHOST'
+Sender_alias = 'Services'
+Subject = 'New User Account'
+Info_url = 'https://www.google.com'
+Mail_list = 'root@localhost'
+Mail_list_bcc = 'cmsupport@localhost'
+Support_email = 'support@listserv.uab.edu'
+
+Head = f"""From: {Sender_alias} <{Sender}>
+To: <{{{{ to }}}}>
+Subject: {Subject}
+"""
+
+Body = f"""
+Hi {{{{ username }}}}
+Your account has been set up with:
+
+============================
+User ID:  {{{{ username }}}}
+============================
+
+If you have any questions, please visit:
+{Info_url}
+
+or email at {Admin_email}
+
+Cheers,
+"""
+
+Whole_mail = Head + Body

rc_rmq.py

@@ -3,26 +3,24 @@ import pika
 import socket
 import rabbit_config as rcfg
 
+
 class RCRMQ(object):
 
-    USER = 'guest'
-    PASSWORD = 'guest'
-    HOST = 'localhost'
+    USER = "guest"
+    PASSWORD = "guest"
+    HOST = "localhost"
     PORT = 5672
-    VHOST = '/'
-    EXCHANGE = ''
-    EXCHANGE_TYPE = 'direct'
-    QUEUE = None
-    DURABLE = True
-    ROUTING_KEY = None
+    VHOST = "/"
+    EXCHANGE = ""
+    EXCHANGE_TYPE = "direct"
     DEBUG = False
 
     def __init__(self, config=None, debug=False):
         if config:
-            if 'exchange' in config:
-                self.EXCHANGE = config['exchange']
-            if 'exchange_type' in config:
-                self.EXCHANGE_TYPE = config['exchange_type']
+            if "exchange" in config:
+                self.EXCHANGE = config["exchange"]
+            if "exchange_type" in config:
+                self.EXCHANGE_TYPE = config["exchange_type"]
 
         hostname = socket.gethostname().split(".", 1)[0]
 
@@ -34,7 +32,8 @@ class RCRMQ(object):
         self.DEBUG = debug
 
         if self.DEBUG:
-            print("""
+            print(
+                """
             Created RabbitMQ instance with:
               Exchange name: {},
               Exchange type: {},
@@ -42,70 +41,104 @@ class RCRMQ(object):
               User: {},
               VHost: {},
               Port: {}
-            """.format(self.EXCHANGE, self.EXCHANGE_TYPE, self.HOST, self.USER, self.VHOST, self.PORT))
+            """.format(
+                    self.EXCHANGE,
+                    self.EXCHANGE_TYPE,
+                    self.HOST,
+                    self.USER,
+                    self.VHOST,
+                    self.PORT,
+                )
+            )
 
         self._consumer_tag = None
         self._connection = None
         self._consuming = False
         self._channel = None
         self._parameters = pika.ConnectionParameters(
-                self.HOST,
-                self.PORT,
-                self.VHOST,
-                pika.PlainCredentials(self.USER, self.PASSWORD))
+            self.HOST,
+            self.PORT,
+            self.VHOST,
+            pika.PlainCredentials(self.USER, self.PASSWORD),
+        )
 
     def connect(self):
         if self.DEBUG:
-            print("Connecting...\n" + "Exchange: " + self.EXCHANGE + " Exchange type: " + self.EXCHANGE_TYPE)
+            print(
+                "Connecting...\n"
+                + "Exchange: "
+                + self.EXCHANGE
+                + " Exchange type: "
+                + self.EXCHANGE_TYPE
+            )
 
         self._connection = pika.BlockingConnection(self._parameters)
         self._channel = self._connection.channel()
         self._channel.exchange_declare(
-                exchange=self.EXCHANGE,
-                exchange_type=self.EXCHANGE_TYPE,
-                durable=True)
+            exchange=self.EXCHANGE,
+            exchange_type=self.EXCHANGE_TYPE,
+            durable=True,
+        )
+
+    def bind_queue(
+        self, queue="", routing_key=None, durable=True, exclusive=False
+    ):
+
+        if self._connection is None:
+            self.connect()
+
+        result = self._channel.queue_declare(
+            queue=queue, durable=durable, exclusive=exclusive
+        )
 
-    def bind_queue(self):
-        self._channel.queue_declare(queue=self.QUEUE, durable=self.DURABLE)
-        self._channel.queue_bind(exchange=self.EXCHANGE,
-                queue=self.QUEUE,
-                routing_key=self.ROUTING_KEY)
+        self._channel.queue_bind(
+            exchange=self.EXCHANGE,
+            queue=result.method.queue,
+            routing_key=routing_key,
+        )
+
+        return result.method.queue
 
     def disconnect(self):
-        self._channel.close()
-        self._connection.close()
-        self._connection = None
+        if self._connection:
+            self._channel.close()
+            self._connection.close()
+            self._connection = None
 
-    def delete_queue(self):
-        self._channel.queue_delete(self.QUEUE)
+    def delete_queue(self, queue):
+        self._channel.queue_delete(queue)
 
     def publish_msg(self, obj):
-        if 'routing_key' in obj:
-            self.ROUTING_KEY = obj['routing_key']
+        routing_key = obj.get("routing_key")
+        props = obj.get("props")
 
         if self._connection is None:
             self.connect()
 
-        self._channel.basic_publish(exchange=self.EXCHANGE,
-                routing_key=self.ROUTING_KEY,
-                body=json.dumps(obj['msg']))
+        self._channel.basic_publish(
+            exchange=self.EXCHANGE,
+            routing_key=routing_key,
+            properties=props,
+            body=json.dumps(obj["msg"]),
+        )
 
     def start_consume(self, obj):
-        if 'queue' in obj:
-            self.QUEUE = obj['queue']
-            self.ROUTING_KEY = obj['routing_key'] if 'routing_key' in obj else self.QUEUE
-        if 'durable' in obj:
-            self.DURABLE = obj['durable']
-
-        if self.DEBUG:
-            print("Queue: " + self.QUEUE + "\nRouting_key: " + self.ROUTING_KEY)
+        queue = obj.get("queue", "")
+        routing_key = obj.get("routing_key", queue or None)
+        durable = obj.get("durable", True)
+        exclusive = obj.get("exclusive", False)
+        bind = obj.get("bind", True)
 
         if self._connection is None:
             self.connect()
 
-        self.bind_queue()
+        if bind:
+            self.bind_queue(queue, routing_key, durable, exclusive)
+
+        if self.DEBUG:
+            print("Queue: " + queue + "\nRouting_key: " + routing_key)
 
-        self._consumer_tag = self._channel.basic_consume(self.QUEUE,obj['cb'])
+        self._consumer_tag = self._channel.basic_consume(queue, obj["cb"])
         self._consuming = True
         try:
             self._channel.start_consuming()

rc_util.py

+import errno
+import functools
+import os
+import signal
 import logging
 import argparse
+import pika
+import pwd
+import uuid
 from rc_rmq import RCRMQ
 import json
+from urllib.parse import quote
+from time import sleep
+import rabbit_config as rcfg
+
+rc_rmq = RCRMQ({"exchange": "RegUsr", "exchange_type": "topic"})
+tasks = {
+    "create_account": None,
+    "git_commit": None,
+    "dir_verify": None,
+    "subscribe_mail_list": None,
+    "notify_user": None,
+}
+logger_fmt = "%(asctime)s [%(module)s] - %(message)s"
+
+
+class TimeoutError(Exception):
+    pass
+
+
+# From https://stackoverflow.com/questions/2281850
+def timeout(seconds=30, error_message=os.strerror(errno.ETIME)):
+    def decorator(func):
+        def _handle_timeout(signum, frame):
+            raise TimeoutError(error_message)
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            signal.signal(signal.SIGALRM, _handle_timeout)
+            signal.alarm(seconds)
+            try:
+                result = func(*args, **kwargs)
+            finally:
+                signal.alarm(0)
+            return result
+
+        return wrapper
+
+    return decorator
+
+
+def add_account(
+    username, queuename, email, full="", reason="", updated_by="", host=""
+):
+    rc_rmq.publish_msg(
+        {
+            "routing_key": "request." + queuename,
+            "msg": {
+                "username": username,
+                "email": email,
+                "fullname": full,
+                "reason": reason,
+                "queuename": queuename,
+                "updated_by": updated_by,
+                "host": host,
+            },
+        }
+    )
+    rc_rmq.disconnect()
+
+
+def certify_account(
+    username, queuename, state="ok", service="all", updated_by="", host=""
+):
+    rc_rmq.publish_msg(
+        {
+            "routing_key": "acctmgr.request." + queuename,
+            "msg": {
+                "username": username,
+                "service": service,
+                "state": state,
+                "queuename": queuename,
+                "updated_by": updated_by,
+                "host": host,
+            },
+        }
+    )
+    rc_rmq.disconnect()
 
-rc_rmq = RCRMQ({'exchange': 'RegUsr', 'exchange_type': 'topic'})
-tasks = {'ohpc_account': None, 'ood_account': None, 'slurm_account': None}
-logger_fmt = '%(asctime)s [%(module)s] - %(message)s'
-
-def add_account(username, email, full='', reason=''):
-  rc_rmq.publish_msg({
-    'routing_key': 'request.' + username,
-    'msg': {
-      "username": username,
-      "email": email,
-      "fullname": full,
-      "reason": reason
-    }
-  })
-  rc_rmq.disconnect()
 
 def worker(ch, method, properties, body):
     msg = json.loads(body)
-    task = msg['task']
-    tasks[task] = msg['success']
-    print("Got msg: {}({})".format(msg['task'], msg['success']))
-
-    # Check if all tasks are done
-    done = True
-    for key, status in tasks.items():
-        if not status:
-            print("{} is not done yet.".format(key))
-            done = False
-    if done:
-        rc_rmq.stop_consume()
-        rc_rmq.delete_queue()
-
-def consume(username, callback=worker, debug=False):
+    username = msg["username"]
+
+    if msg["success"]:
+        print(f"Account for {username} has been created.")
+    else:
+        print(f"There's some issue while creating account for {username}")
+        errmsg = msg.get("errmsg", [])
+        for err in errmsg:
+            print(err)
+
+    rc_rmq.stop_consume()
+    rc_rmq.delete_queue()
+
+
+def consume(
+    queuename,
+    routing_key="",
+    callback=worker,
+    bind=True,
+    durable=True,
+    exclusive=False,
+    debug=False,
+):
+    if routing_key == "":
+        routing_key = "complete." + queuename
+
     if debug:
         sleep(5)
     else:
-        rc_rmq.start_consume({
-            'queue': username,
-            'routing_key': 'confirm.' + username,
-            'cb': callback
-        })
+        rc_rmq.start_consume(
+            {
+                "queue": queuename,
+                "routing_key": routing_key,
+                "bind": bind,
+                "durable": durable,
+                "exclusive": exclusive,
+                "cb": callback,
+            }
+        )
         rc_rmq.disconnect()
 
-    return { 'success' : True }
+    return {"success": True}
+
 
 def get_args():
     # Parse arguments
     parser = argparse.ArgumentParser()
-    parser.add_argument('-v', '--verbose', action='store_true', help='verbose output')
-    parser.add_argument('-n', '--dry-run', action='store_true', help='enable dry run mode')
+    parser.add_argument(
+        "-v", "--verbose", action="store_true", help="verbose output"
+    )
+    parser.add_argument(
+        "-n", "--dry-run", action="store_true", help="enable dry run mode"
+    )
     return parser.parse_args()
 
+
 def get_logger(args=None):
     if args is None:
         args = get_args()
@@ -70,3 +161,125 @@ def get_logger(args=None):
     logging.basicConfig(format=logger_fmt, level=logger_lvl)
     return logging.getLogger(__name__)
 
+
+def encode_name(uname):
+    uname_quote = quote(uname)
+    if "." in uname_quote:
+        uname_quote = uname_quote.replace(".", "%2E")
+    return uname_quote
+
+
+@timeout(rcfg.Function_timeout)
+def check_state(username, debug=False):
+    corr_id = str(uuid.uuid4())
+    result = ""
+    rpc_queue = "user_state"
+
+    def handler(ch, method, properties, body):
+        if debug:
+            print("Message received:")
+            print(body)
+
+        nonlocal corr_id
+        nonlocal result
+        msg = json.loads(body)
+
+        if corr_id == properties.correlation_id:
+            if not msg["success"]:
+                print("Something's wrong, please try again.")
+            else:
+                result = msg.get("state")
+
+            rc_rmq.stop_consume()
+            rc_rmq.disconnect()
+
+    callback_queue = rc_rmq.bind_queue(exclusive=True)
+
+    if debug:
+        print(f"Checking state of user {username}")
+        print(f"Callback queue: {callback_queue}, correlation_id: {corr_id}")
+
+    rc_rmq.publish_msg(
+        {
+            "routing_key": rpc_queue,
+            "props": pika.BasicProperties(
+                correlation_id=corr_id, reply_to=callback_queue
+            ),
+            "msg": {"op": "get", "username": username},
+        }
+    )
+
+    rc_rmq.start_consume(
+        {
+            "queue": callback_queue,
+            "exclusive": True,
+            "bind": False,
+            "cb": handler,
+        }
+    )
+
+    return result
+
+
+@timeout(rcfg.Function_timeout)
+def update_state(username, state, updated_by="", host="", debug=False):
+
+    if state not in rcfg.Valid_state:
+        print(f"Invalid state '{state}'")
+        return False
+
+    corr_id = str(uuid.uuid4())
+    result = ""
+    rpc_queue = "user_state"
+
+    def handler(ch, method, properties, body):
+        if debug:
+            print("Message received:")
+            print(body)
+
+        nonlocal corr_id
+        nonlocal result
+        msg = json.loads(body)
+
+        if corr_id == properties.correlation_id:
+            if not msg["success"]:
+                print("Something's wrong, please try again.")
+
+            result = msg["success"]
+            rc_rmq.stop_consume()
+            rc_rmq.disconnect()
+
+    callback_queue = rc_rmq.bind_queue(exclusive=True)
+
+    rc_rmq.publish_msg(
+        {
+            "routing_key": rpc_queue,
+            "props": pika.BasicProperties(
+                reply_to=callback_queue, correlation_id=corr_id
+            ),
+            "msg": {
+                "op": "post",
+                "username": username,
+                "state": state,
+                "updated_by": updated_by,
+                "host": host,
+            },
+        }
+    )
+
+    rc_rmq.start_consume(
+        {
+            "queue": callback_queue,
+            "exclusive": True,
+            "bind": False,
+            "cb": handler,
+        }
+    )
+
+    return result
+
+
+def get_caller_info():
+    username = pwd.getpwuid(os.getuid()).pw_name
+    hostname = os.uname().nodename
+    return (username, hostname)

requirements.txt

 pika==1.1.0
+dataset==1.3.1
+Jinja2==2.11.2
+sh==1.12.14
+pre-commit==2.12.1
+greenlet==1.1.3

scripts/crondisable.sh

+#!/bin/bash
+
+usage() {
+    echo "Usage: $0 -d USERNAME to disable a user"
+    echo "Usage: $0 -e USERNAME to re-enable a user"
+}
+
+if [[ "$EUID" -ne 0 ]]; then
+    echo "This script must be run as root!"
+    exit 1
+fi
+
+if [ "$#" -eq 0 ]; then
+    usage
+    exit 1
+fi
+
+while getopts ':d:e:' OPTION; do
+    case $OPTION in
+    d)
+        mv /var/spool/cron/$OPTARG /var/spool/cron/$OPTARG.disabled
+        echo $OPTARG >>/etc/cron.deny
+        ;;
+    e)
+        mv /var/spool/cron/$OPTARG.disabled /var/spool/cron/$OPTARG
+        sed -i -e "/$OPTARG/d" /etc/cron.deny
+        ;;
+    esac
+done

scripts/kill_user_process.sh

+#!/bin/bash
+
+force=false
+username=''
+
+usage() {
+  echo "Usage: $0 -u USERNAME to run graceful shutdown of processes"
+  echo "Usage: $0 -u USERNAME -k to run forced shutdown of processes"
+}
+
+if [[ "$EUID" -ne 0 ]]; then
+  echo "This script must be run as root!"
+  exit 1
+fi
+
+while [ $OPTIND -le "$#" ]; do
+  if getopts k option; then
+    case $option in
+    k)
+      force=true
+      ;;
+    esac
+  else
+    username=("${!OPTIND}")
+    ((OPTIND++))
+  fi
+done
+
+if [ -z "$username" ]; then
+  usage
+  exit 1
+fi
+
+if [ "$username" = "root" ]; then
+  echo "Username cannot be root"
+  exit 1
+fi
+
+userId=$(id -u $username)
+
+if [ "$force" = true ]; then
+  echo "Performing SIGKILL on processes belonging to $username"
+  pkill -9 -u $userId
+else
+  echo "Performing SIGTERM on processes belonging to $username"
+  pkill -u $userId
+fi

user_auth.py

+#!/usr/bin/env python3
+import sys
+import rc_util
+import subprocess
+
+# During migration of this new script for ood
+# e.g. not all of users are in the db
+migration = True
+default_state = "ok"
+# migration = False  # uncomment after migration's done
+remote_user = sys.argv[1]
+
+result = rc_util.check_state(remote_user)
+
+if result == "ok":
+    print(remote_user)
+else:
+    if migration and result == "no-account":
+        rc = subprocess.run(
+            ["getent", "passwd", remote_user], stdout=subprocess.DEVNULL
+        ).returncode
+        if rc == 0:
+            rc_util.update_state(remote_user, default_state)
+            print(remote_user)
+            sys.exit()
+    print()

user_migrate.sh

+#!/bin/bash
+
+group_options=(gpfs4 gpfs5)
+gpfs4_home="/gpfs4/data/user/home"
+gpfs5_home="/gpfs5/data/user/home"
+
+user=$1
+group_to=$2
+
+if [[ -z "${group_to}" ]]; then
+  echo "Usage: $0 USER TARGET_GROUP"
+  exit 1
+elif [[ ! " ${group_options[*]} " =~ [[:space:]]${group_to}[[:space:]] ]]; then
+  echo "Invalid target group"
+  echo "Available options: ${group_options[*]}, got ${group_to}"
+  exit 1
+fi
+
+if ! getent passwd "$user" > /dev/null 2>&1; then
+  echo "The user $user does not exist"
+  exit 1
+fi
+
+cd /cm/shared/rabbitmq_agents || exit
+source venv/bin/activate
+
+if [[ "$group_to" == "gpfs4" ]]; then
+  group_from=gpfs5
+  dir_from="$gpfs5_home/$user/"
+  dir_to="$gpfs4_home/$user"
+else
+  group_from=gpfs4
+  dir_from="$gpfs4_home/$user/"
+  dir_to="$gpfs5_home/$user"
+fi
+
+if [[ -d "/$group_from/data/user/home/$user" ]]; then
+  ./account_manager.py "$user" hold
+
+  rsync -a --delete "$dir_from" "$dir_to"
+
+  ./group_manager.py "$user" -g "$group_to"
+  ./group_manager.py "$user" -d -g "$group_from"
+
+  ./account_manager.py "$user" ok
+else
+  echo User home directory does not exist.
+  exit 1
+fi