petarnikolovski · December 7, 2018 21:40 · Dec 7, 2018
diff --git a/users.py b/users.py
@@ -0,0 +1,206 @@
+import sqlite3
+from pathlib import Path
+from functools import wraps
+
+
+"""
+This is a dummy implementation of the privilege-based checks based on the following article:
+https://lostechies.com/derickbailey/2011/05/24/dont-do-role-based-authorization-checks-do-activity-based-checks/
+"""
+
+
+class Settings:
+
+    db = 'users.sqlite3'
+
+
+def initiate_database():
+    if Path(Settings.db).exists():
+        return
+
+    conn = sqlite3.connect(Settings.db)
+    c = conn.cursor()
+
+    c.execute("""
+    CREATE TABLE roles (
+        id INTEGER,
+        role TEXT
+    )
+    """)
+
+    c.execute("""
+    CREATE TABLE privileges (
+        id INTEGER,
+        privilege TEXT
+    )
+    """)
+
+    c.execute("""
+    CREATE TABLE roles_privileges (
+        id INTEGER,
+        id_role INTEGER,
+        id_privilege INTEGER,
+        FOREIGN KEY(id_role) REFERENCES roles(id),
+        FOREIGN KEY(id_privilege) REFERENCES privileges(id)
+    )
+    """)
+
+    c.execute("""
+    CREATE TABLE users (
+        id INTEGER,
+        username TEXT,
+        password TEXT,
+        id_role INTEGER,
+        FOREIGN KEY(id_role) REFERENCES roles(id)
+    )
+    """)
+
+    c.execute("INSERT INTO roles VALUES (1, 'teacher')")
+    c.execute("INSERT INTO roles VALUES (2, 'student')")
+
+    c.execute("INSERT INTO privileges VALUES (1, 'create')")
+    c.execute("INSERT INTO privileges VALUES (2, 'read')")
+
+    c.execute("INSERT INTO roles_privileges VALUES (1, 1, 1)")
+    c.execute("INSERT INTO roles_privileges VALUES (1, 1, 2)")
+    c.execute("INSERT INTO roles_privileges VALUES (2, 2, 2)")
+
+    c.execute("INSERT INTO users VALUES (1, 'john', 'password', 1)")
+    c.execute("INSERT INTO users VALUES (2, 'jane', 'password', 2)")
+
+    conn.commit()
+    conn.close()
+
+
+class Privileges:
+
+    CREATE = 'create'
+    READ = 'read'
+
+
+class Roles:
+
+    TEACHER = 'teacher'
+    STUDENT = 'student'
+
+
+class Db:
+
+    def __init__(self, filename):
+        self.filename = filename
+
+    def select_many(self, statement, *args):
+        conn = sqlite3.connect(Settings.db)
+        c = conn.cursor()
+
+        c.execute(statement, *args)
+        result = c.fetchall()
+
+        conn.close()
+        return result
+
+    def select_one(self, statement, *args):
+        conn = sqlite3.connect(Settings.db)
+        c = conn.cursor()
+
+        c.execute(statement, args)
+        result = c.fetchone()
+
+        conn.close()
+        return result
+
+
+class RoleRepository:
+
+    def __init__(self):
+        self.db = Db(Settings.db)
+
+    def get_roles_for_activity(self, activity):
+        sql_activity = 'SELECT id FROM privileges WHERE privilege=?'
+        activity_id = self.db.select_one(sql_activity, activity)
+
+        sql = 'SELECT id_role FROM roles_privileges WHERE id_privilege=?'
+        role_ids = self.db.select_many(sql, activity_id)
+
+        sql_roles = 'SELECT role FROM roles WHERE id=?'
+        roles = []
+        for id in role_ids:
+            role = self.db.select_one(sql_roles, *id)
+            roles.append(role)
+
+        return roles
+
+
+def authorize_activity(activity=''):
+    if not activity:
+        raise Exception('You MUST use this decorator with activity paramater')
+
+    def authorize(fun):
+        if not fun:
+            raise Exception('Use this decorator with activity paramater')
+
+        @wraps(fun)
+        def wrapper(*args, **kwargs):
+            user = LoginMiddleware.user
+            if not user:
+                raise Exception('Session expired!')
+
+            role_repository = RoleRepository()
+            privileged = role_repository.get_roles_for_activity(activity)
+            if user.role in [p[0] for p in privileged]:
+                fun(*args, **kwargs)
+            else:
+                e = '{} does not have permission!'.format(user.username)
+                raise Exception(e)
+
+            return
+        return wrapper
+    return authorize
+
+
+class User:
+
+    def __init__(self, username, password, role):
+        self.username = username
+        self.password = password
+        self.role = role
+
+
+class LoginMiddleware:
+
+    user = None
+
+    @classmethod
+    def login(cls, username, password):
+        db = Db(Settings.db)
+        sql = 'SELECT id_role FROM users WHERE username=? AND password=?'
+        user = db.select_one(sql, username, password)
+        if not user:
+            raise Exception('User not found!')
+        sql_role = 'SELECT role FROM roles WHERE id=?'
+        role = db.select_one(sql_role, *user)
+        cls.user = User(username, password, *role)
+        return cls.user
+
+
+class LMS:
+
+    @authorize_activity(activity=Privileges.CREATE)
+    def create_student(self):
+        print('It creates.')
+
+    @authorize_activity(activity=Privileges.READ)
+    def view_lesson(self):
+        print('It reads.')
+
+
+if __name__ == '__main__':
+    initiate_database()
+
+    username = input('Please enter your username: ')
+    password = input('Please enter your password: ')
+
+    user = LoginMiddleware.login(username, password)
+
+    LMS().view_lesson()
+    LMS().create_student()
No results found