Continuous assessment 02 - Protection of information based on sensitivity and privilege levels

config.csv

constants.py

# file names

CONFIG_FILE = 'config.csv'
DATA_FILE = 'data.csv'

# data fields

USERNAME = 'username'
PASSWORD = 'password'
USER_TYPE = 'user_type'
PRIVILEGE_LEVEL = 'privilege_level'

ID = 'id'
PERSONAL_DETAILS = 'personal_details'
SICKNESS_DETAILS = 'sickness_details'
DRUG_PRESCRIPTIONS = 'drug_prescriptions'
LAB_TEST_PRESCRIPTIONS = 'lab_test_prescriptions'


# user types

PATIENT = 'patient'
HOSPITAL_STAFF = 'hospital_staff'

LAB_STAFF = 'lab_staff'
PHARMACY_STAFF = 'pharmacy_staff'
NURSE = 'nurse'
DOCTOR = 'doctor'

# privilege levels

L1 = 'L1'                        # patient
L2 = 'L2'                        # lab staff
L3 = 'L3'                        # pharmacy staff
L4 = 'L4'                        # nurse
L5 = 'L5'                        # doctor

# actions

VIEW_MY_RECORDS = 'view my records'
VIEW_ALL_RECORDS = 'view all records'
VIEW_RECORDS_BY_USER = 'view records by user'
VIEW_RECORD_BY_ID = 'view record by id'
ADD_RECORD = 'add record'
EDIT_RECORD = 'edit record'

data.csv

medical_data_processor.py

import hashlib
import os.path
import shutil
from tempfile import NamedTemporaryFile
from csv import DictWriter, DictReader

# constants used to minimize mistakes and code completion
from constants import CONFIG_FILE, DATA_FILE,\
    USERNAME, PASSWORD, USER_TYPE, PRIVILEGE_LEVEL, \
    ID, PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS, \
    PATIENT, HOSPITAL_STAFF, LAB_STAFF, PHARMACY_STAFF, NURSE, DOCTOR, \
    L1, L2, L3, L4, L5,\
    VIEW_MY_RECORDS, VIEW_RECORDS_BY_USER, VIEW_ALL_RECORDS, VIEW_RECORD_BY_ID, ADD_RECORD, EDIT_RECORD

# file names
config_file = CONFIG_FILE
config_fields = [USERNAME, PASSWORD, USER_TYPE, PRIVILEGE_LEVEL]

data_file = DATA_FILE
data_fields = [ID, USERNAME, PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS]

tempfile = NamedTemporaryFile(mode='w+t',delete=False,newline='')  # temp file for editing a row in data.csv

# user types
user_types = [PATIENT, HOSPITAL_STAFF]
privilege_levels = [L1, L2, L3, L4, L5]

# access control list
acl = {
    L1: {
        'user': PATIENT,
        'read': [ID, USERNAME, PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS],
        'write': [],
        'edit': [],
        'actions': [VIEW_MY_RECORDS]
    },
    L2: {
        'user': LAB_STAFF,
        'read': [ID, USERNAME, LAB_TEST_PRESCRIPTIONS],
        'write': [],
        'edit': [],
        'actions': [VIEW_ALL_RECORDS, VIEW_RECORD_BY_ID, VIEW_RECORDS_BY_USER]
    },
    L3: {
        'user': PHARMACY_STAFF,
        'read': [ID, USERNAME, DRUG_PRESCRIPTIONS],
        'write': [],
        'edit': [],
        'actions': [VIEW_ALL_RECORDS, VIEW_RECORD_BY_ID, VIEW_RECORDS_BY_USER]
    },
    L4: {
        'user': NURSE,
        'read': [ID, USERNAME, PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS],
        'write': [USERNAME, PERSONAL_DETAILS],
        'edit': [PERSONAL_DETAILS],
        'actions': [VIEW_ALL_RECORDS, VIEW_RECORDS_BY_USER, VIEW_RECORD_BY_ID, ADD_RECORD, EDIT_RECORD]
    },
    L5: {
        'user': DOCTOR,
        'read': [ID, USERNAME, PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS],
        'write': [USERNAME, PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS],
        'edit': [PERSONAL_DETAILS, SICKNESS_DETAILS, DRUG_PRESCRIPTIONS, LAB_TEST_PRESCRIPTIONS],
        'actions': [VIEW_ALL_RECORDS, VIEW_RECORDS_BY_USER, VIEW_RECORD_BY_ID, ADD_RECORD, EDIT_RECORD]
    }
}


"""
-------
begin - Helper functions
-------
"""
def hash_password(password):              # hash string password into md5
    return hashlib.md5(password.encode()).hexdigest()

def append_row_to_csv(file_name, data_record, fields):  # append a data row to csv file
    file_exists = os.path.isfile(file_name)
    with open(file_name, 'a+', newline='') as csv_file:
        writer = DictWriter(csv_file, fieldnames=fields)
        if not file_exists:
            writer.writeheader()
        writer.writerow(data_record)

def check_username_exists(username):           # check if username already exists
    file_exists = os.path.isfile(config_file)
    if file_exists:
        with open(config_file, newline='') as config:
            reader = DictReader(config)
            for row in reader:
                if row[USERNAME].casefold() == username.strip().casefold():   # case insensitive, trim whitespaces
                    return True
    return False

def check_user_exists_and_patient(username):           # check if user exists and is a patient
    file_exists = os.path.isfile(config_file)
    if file_exists:
        with open(config_file, newline='') as config:
            reader = DictReader(config)
            for row in reader:
                if row[USERNAME].casefold() == username.strip().casefold() and row[USER_TYPE] == PATIENT:
                    return True
    return False

def check_string_empty(string):  # check if string is empty
    return string.strip() == ''

def check_comma_in_string(string):      # check if comma exist in a string
    return ',' in string

def check_spaces_in_string(string):      # check if comma exist in a string
    trimmed_str = string.strip()
    return ' ' in trimmed_str

def check_username_valid(username):        # check if username is valid
    if check_string_empty(username):
        return False
    if check_comma_in_string(username):
        return False
    if check_spaces_in_string(username):
        return False
    if check_username_exists(username):
        return False
    return True

def check_password_valid(password):        # check if password is valid
    if check_string_empty(password):
        return False
    return True

def get_autoincrement_id():       # get next record id
    autoincrement_id = 0
    file_exists = os.path.isfile(data_file)
    if file_exists:
        with open(data_file, newline='') as data:
            reader = DictReader(data)
            for row in reader:
                autoincrement_id = int(row[ID])
    return autoincrement_id+1

def get_privilege_level(user_type):        # return privilege level
    for i in acl:
        if acl[i]['user'] == user_type:
            return i
    return 'Invalid user type'
"""
-------
end - Helper functions
-------
"""

"""
-------
begin - Action functions
-------
"""
# view the values related to the given input field of all records in data.csv
def view_all_records(fields_list):
    data_found = False
    file_exists = os.path.isfile(data_file)
    if file_exists:
        with open(data_file, newline='') as data:
            reader = DictReader(data)
            for row in reader:
                data_found = True
                print('\n')
                for field in fields_list:
                    print(str(field) + ' : ' + str(row[field]))
    if not data_found:
        print('Data records not found!!')

# view the values related to the given input field of given user in data.csv
def view_records_by_user(fields_list, username):
    data_found = False
    file_exists = os.path.isfile(data_file)
    if file_exists:
        with open(data_file, newline='') as data:
            reader = DictReader(data)
            for row in reader:
                if row[USERNAME] == username:
                    data_found = True
                    print('\n')
                    for field in fields_list:
                        print(str(field) + ' : ' + str(row[field]))
    if not data_found:
        print('Data records for user not found!!')

# view the values related to the given input field of given record id in data.csv
def view_records_by_id(fields_list, id):
    data_found = False
    file_exists = os.path.isfile(data_file)
    if file_exists:
        with open(data_file, newline='') as data:
            reader = DictReader(data)
            for row in reader:
                if row[ID] == id:
                    data_found = True
                    print('\n')
                    for field in fields_list:
                        print(str(field) + ' : ' + str(row[field]))
    if not data_found:
        print('Data record for ID not found!!')

# add a new record (values for only given fields) to data.csv
def add_record(fields_list):
    new_record = {
        ID: get_autoincrement_id(),
        USERNAME: '',
        PERSONAL_DETAILS: '',
        SICKNESS_DETAILS: '',
        DRUG_PRESCRIPTIONS: '',
        LAB_TEST_PRESCRIPTIONS: ''
    }
    for field in fields_list:

        if field == USERNAME:
            username_exists_and_patient = False
            while not username_exists_and_patient:
                username = input('Enter username : ')
                username_exists_and_patient = check_user_exists_and_patient(username)
                if not username_exists_and_patient:
                    print('Username does not exist or not a patient!! \n')
                new_record[USERNAME] = username
        else:
            comma_in_input = True
            while comma_in_input:
                user_input = input('Enter ' + field + ' : ')
                comma_in_input = check_comma_in_string(user_input)
                if comma_in_input:
                    print('Cannot have commas in input!!')
                new_record[field] = user_input

    append_row_to_csv(data_file, new_record, data_fields)
    print('New record added!!')

# edit an existing record (values for only given fields) in data.csv
def edit_record(id, fields_list):
    edit_choice = ''
    file_exists = os.path.isfile(data_file)
    if file_exists:
        with open(data_file, 'rt', newline='') as data, tempfile:
            reader = DictReader(data)
            writer = DictWriter(tempfile, fieldnames=data_fields)
            writer.writeheader()
            for row in reader:
                if row[ID] == id:
                    for field in fields_list:
                        valid_edit_choice = False
                        while not valid_edit_choice:
                            edit_choice = input('Do you want to edit '+ field + ' ? (y/n) : ')
                            if edit_choice in ['y', 'n']:
                                valid_edit_choice = True

                        if edit_choice == 'y':
                            comma_in_input = True
                            while comma_in_input:
                                user_input = input('Enter ' + field + ' : ')
                                comma_in_input = check_comma_in_string(user_input)
                                if comma_in_input:
                                    print('Cannot have commas in input!! \n')
                                row[field] = user_input
                        else:
                            continue
                writer.writerow(row)
        shutil.move(tempfile.name, data_file)
        print('Record successfully edited!!')
    else:
        print('No record to edit!!')
"""
-------
end - Action functions
-------
"""

def action_selector(action, session):       # select the action to perform
    if action == VIEW_MY_RECORDS:
        view_records_by_user(acl[session[PRIVILEGE_LEVEL]]['read'], session[USERNAME])
    elif action == VIEW_ALL_RECORDS:
        view_all_records(acl[session[PRIVILEGE_LEVEL]]['read'])
    elif action == VIEW_RECORDS_BY_USER:
        input_user = input('Enter a username : ')
        view_records_by_user(acl[session[PRIVILEGE_LEVEL]]['read'], input_user)
    elif action == VIEW_RECORD_BY_ID:
        input_id = input('Enter a record ID : ')
        view_records_by_id(acl[session[PRIVILEGE_LEVEL]]['read'], input_id)
    elif action == ADD_RECORD:
        add_record(acl[session[PRIVILEGE_LEVEL]]['write'])
    elif action == EDIT_RECORD:
        input_id = input('Enter record ID of what you want to edit : ')
        edit_record(input_id,acl[session[PRIVILEGE_LEVEL]]['edit'])
    else:
        print('Action not available')

# login to the system
def login():
    print('-----Login to the system-----')
    username = input('Enter username : ')
    password = input('Enter your password : ')

    with open(config_file, newline='') as config:
        config_rows = DictReader(config)

        for row in config_rows:
            if row[USERNAME] == username:
                if row[PASSWORD] == hash_password(password):
                    return {
                        USERNAME: username,
                        USER_TYPE: row[USER_TYPE],
                        PRIVILEGE_LEVEL: row[PRIVILEGE_LEVEL],
                        'message': 'Success'
                    }
                print('incorrect username or password')
                return {'message': 'Fail'}
        print('incorrect username or password')
        return {'message': 'Fail'}

# register new user to the system
def register():
    user_type, username, privilege_level, password = '', '', '', ''
    print('-----Register to the system-----')
    username_valid = False        # setting username
    while not username_valid:
        username = input('Enter a username : ')
        username_valid = check_username_valid(username)
        if not username_valid:
            print('Enter valid username!! \n')

    password_valid = False       # setting password
    while not password_valid:
        password = input('Enter a password : ')
        password_valid = check_password_valid(password)
        if not password_valid:
            print('Enter valid password!! \n')

    user_type_valid = False         # setting user type
    while not user_type_valid:
        user_type_no = input('Select user type \n 1 - Patient\n 2 - Hospital Staff\n')
        if user_type_no in ['1', '2']:
            user_type = user_types[int(user_type_no)-1]
            user_type_valid = True
        else:
            print('Select valid user type!! \n')

    if user_type != PATIENT:       # setting a privilege level
        staff_type_valid = False
        while not staff_type_valid:
            staff_type_no = input('Select hospital staff type \n 1 - Lab Staff\n 2 - Pharmacy Staff\n 3 - Nurse\n 4 - Doctor\n')
            if staff_type_no in ['1', '2', '3', '4']:
                privilege_level = privilege_levels[int(staff_type_no)]
                staff_type_valid = True
            else:
                print('Select valid hospital staff type!! \n')
    else:
        privilege_level = L1

    config_record = {    # record to add to config file
        USERNAME: username,
        PASSWORD: hash_password(password),     # hashing
        USER_TYPE: user_type,
        PRIVILEGE_LEVEL: privilege_level
    }

    append_row_to_csv(config_file, config_record, config_fields)
    print('User registration successful!!')


# main program
init, user_session, selected_action = '', {}, ''
init_selected = False
while not init_selected:        # prompt to select login or register
    init = input('Login or Register \n 1 - Login\n 2 - Register\n')
    if init in ['1', '2']:
        init_selected = True
    else:
        print('Select valid option!! \n')

if init == '1':
    user_session = login()
    if user_session['message'] == 'Success':  # if logged in
        print('Successfully logged In \n')
        action_string = 'Select actions \n '
        for i in range(len(acl[user_session[PRIVILEGE_LEVEL]]['actions'])):
            action_string += str(i) + ' - ' + acl[user_session[PRIVILEGE_LEVEL]]['actions'][i] + '\n '

        action_selected = False
        while not action_selected:
            selected_action = input(action_string)  # prompt to select from user's available actions
            if int(selected_action) in range(len(acl[user_session[PRIVILEGE_LEVEL]]['actions'])):
                action_selected = True
            else:
                print('Select valid action!! \n')

        if action_selected:
            action_selector(acl[user_session[PRIVILEGE_LEVEL]]['actions'][int(selected_action)], user_session)
else:
    register()