fabaccess/fab_access/main.py

from keycloak import KeycloakAdmin
import json
import psycopg
import os
from paho.mqtt.client import Client as MQTTClient

import config
from mqtt_helper import MQTTHelper

conn: psycopg.Connection
keycloak_admin: KeycloakAdmin

def publish(client, topic,msg):
        result = client.publish(topic, msg)
        status: 0 | 1 = result[0]
        if status == 0:
            #print(f"Send `{msg}` to topic `{topic}`")
            pass
        else:
            print(f"Failed to send message to topic {topic}")

def changePlug(client: MQTTClient, plug_id: str, is_running: bool):
    publish(client,f"/FabLogging/{plug_id}/POWER", is_running)
    cmd = "On" if is_running else "Off"
    publish(client, f"/aktoren/{plug_id}/cmnd/POWER", cmd)

def changeDisplay(client: MQTTClient, reader_id: str, status: int, fab_card_id: str):
    publish(client, f"/cmnd/reader/{reader_id}", '{"Cmd": "message", "MssgID": %s , "ClrTxt":"" , "AddnTxt":"%s"}' % (status,  fab_card_id))

def hasPermission(UserPermsJSON, PermissionPath):
    # ToDo: refactor this
    permission = False
    error = 0
    MachinePermArray = PermissionPath.split(".")
    for UserPerm in UserPermsJSON:
        #print(f"check {UserPerm}")
        UserPermArray = UserPerm.split(".")
        x = 0
        for UserPermPart in UserPermArray:
            if(error == 0):
                #print(f"Compare {UserPermPart} and {MachinePermArray[x]}")
                if not (MachinePermArray[x] == UserPermPart):
                    if(UserPermPart == "*"):
                        #print("* regelt")
                        permission = True
                    else:
                        error = 1
                        #print(f"MISmatch between {MachinePermArray[x]} and {UserPermPart}")
                else:
                    pass
                    #print(f"Match between {MachinePermArray[x]} and {UserPermPart}")
                x = x + 1
        if(error == 1):
            pass
            #print("Error")
        else:
            #print("Hurra")
            permission = True
        error = 0
    return permission

def handle_msg(client: MQTTClient, userdata, msg):
    global conn

    payload = msg.payload.decode()
    print(f"Received `{payload}` from `{msg.topic}` topic")

    fab_card_id = json.loads(msg.payload.decode())["UID"]
    reader_id = msg.topic.split("/")[-1]

    # check user exists
    users = keycloak_admin.get_users({ 'attributes': { 'FabCard': fab_card_id } })
    if (len(users) != 1):
        print(f'Found {len(users)} users with {fab_card_id=}')
        changeDisplay(client, reader_id, 16, fab_card_id)
        return

    # retrieve user attributes from DB
    user = users[0]
    print(f"FabCard matches with user {user['username']}")
    with conn.cursor() as cursor:
        cursor.execute(f'SELECT PlugName,PermissionPath,Status,LastUser FROM ReaderPlug WHERE ReaderID=?;', (reader_id,))
        res = cursor.fetchone()
        if res is None:
            print(f'Error fetching card info from db for {reader_id=}')
            return
        plug_id, permission_path, is_running, last_user = res

    # check permissions
    user_permissions = user['attributes']['FabPermissions'][0]
    if not hasPermission(user_permissions, permission_path):
        changeDisplay(client, reader_id, 7, "")
        return

    # check for mentor
    if is_running and last_user != user['username']:
        try:
            if (keycloak_admin.get_user_groups(user_id=user['id'])[0]['name'] == 'Mentoren'):
                print('Overrided becouse of "Mentor" Group')
            else:
                print(f'Bereits benutzt von {last_user}')
                error = True
        except IndexError:
            print("No groups available")
            error = True
        if(error):
            changeDisplay(client, reader_id, 9, last_user)

    # toggle machine state
    print(f"Turn Plug {'off' if is_running else 'on'}")
    changePlug(client, plug_id, is_running)
    publish(client, f"/FabLogging/{plug_id}/USER", user['username'])

    # ToDo: refactor display name construction
    try:
        firstCombo = user['firstName']+" "
        DisplayUser = firstCombo+user['lastName']
        if(len(firstCombo) >= 7):
            DisplayUser = user['firstName'][:7]
    except KeyError:
        DisplayUser = user['username']
    if(len(DisplayUser) > 7):
        DisplayUser = DisplayUser[0:7] + "." + DisplayUser[7+1: ]
    DisplayUser = DisplayUser[:8]

    changeDisplay(client, reader_id, 20, f"Login\n{DisplayUser}" if is_running else "Bitte anmelden")

    # write new status to db
    with conn.cursor() as cursor:
        cursor.execute(
            f'UPDATE ReaderPlug SET Status=?, LastUser="?" WHERE ReaderID="?";',
            (is_running, user['username'], reader_id),
        )
        conn.commit()

def main():
    global conn

    mqtt_client = MQTTHelper(**config.get_mqtt_config())
    mqtt_client.subscribe("/rfid_reader/#", handle_msg)
    mqtt_client.loop_forever()

    try:
        conn = psycopg.connect(**config.get_database_config())
    except mariadb.Error as e:
        print(f"Error connecting to MariaDB Platform")
        raise e

    keycloak_admin = KeycloakAdmin(**config.get_keycloak_config(), verify=True)


if __name__ == '__main__':
    main()