Initial Commit für das Grundgerüst der Anwendung, bestehend aus der Dev-Container Konfiguration, der Docker und docker-Compose Definitionen sowie der Verzeichnisse App_Server für die Backend-Programme sowie Client für die browserseitigen GUI-Komponenten
This commit is contained in:
1 parent
10549934ea
commit
1ba78992a6
63 files changed
+5410
No files matched your search
@@ -0,0 +1,258 @@
|
||||
import sqlite3
|
||||
from typing import List
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy import ForeignKey, event
|
||||
from sqlalchemy import Column, Integer, String, DateTime
|
||||
import datetime, decimal
|
||||
from db import dbauth
|
||||
|
||||
|
||||
SYS_ADMIN = "SysAdmin"
|
||||
APP_ADMIN = "AppAdmin"
|
||||
APP_USER = "User"
|
||||
APP_DOZENT = "Doz"
|
||||
|
||||
BINDKEY = 'auth'
|
||||
#c = Connection()
|
||||
|
||||
def columns2dict(record, excludefields=[]):
|
||||
if not isinstance(excludefields, list):
|
||||
excludefields = []
|
||||
attribs = dict()
|
||||
if hasattr(record, '__table__'):
|
||||
if hasattr(record.__table__, 'columns'):
|
||||
for col in record.__table__.columns:
|
||||
if col.name in excludefields:
|
||||
continue
|
||||
value = getattr(record, col.name)
|
||||
if isinstance(value, datetime.date):
|
||||
value = value.isoformat()
|
||||
if isinstance(value, decimal.Decimal):
|
||||
value = "{0:.2f}".format(value)
|
||||
attribs[col.name] = value
|
||||
return attribs
|
||||
|
||||
@event.listens_for(Engine, "connect")
|
||||
def set_sqlite_pragma(dbapi_connection, connection_record):
|
||||
# Dieser Trigger darf nur bei einer Sqlite-Verbindung zünden.
|
||||
# Bei einer VErbindugn zum MS SQL-Server darf er nicht ausgelöst werden
|
||||
if isinstance(dbapi_connection, sqlite3.Connection):
|
||||
cursor = dbapi_connection.cursor()
|
||||
cursor.execute("PRAGMA foreign_keys=ON")
|
||||
cursor.close()
|
||||
|
||||
class Systeminfo(dbauth.Model):
|
||||
__tablename__ = 'systeminfo'
|
||||
__bind_key__ = BINDKEY
|
||||
id = Column(Integer, primary_key=True)
|
||||
version = Column(String(20), nullable=False)
|
||||
lastupdate = Column(DateTime, nullable=True)
|
||||
initialinstall = Column(DateTime, nullable=True)
|
||||
|
||||
# diese Tabelle enthält für alle Apps diejenigen Rollen, die in der App notwendig sind
|
||||
# Da die Rollen nur einmal in der Tabelle "Role" definiert sind und mehrere Apps die gleichen
|
||||
# Rollen haben können, handelt es sich im eine "many-to-many"-Relation, die in dieser
|
||||
# Hilfstabelle verwaltet werden
|
||||
apps_roles = dbauth.Table(
|
||||
'apps_roles',
|
||||
Column('appid', Integer, ForeignKey('applications.id'), primary_key=True),
|
||||
Column('roleid', Integer, ForeignKey('roles.id'), primary_key=True),
|
||||
bind_key=BINDKEY
|
||||
)
|
||||
|
||||
# diese Tabelle enthält für alle Apps, in denen der User eine Rolle hat, die Rollen,
|
||||
# die der User in der jeweiligen App hat.
|
||||
# Da die Rollen nur einmal in der Tabelle "Role" definiert sind und verschiedene einem User
|
||||
# zugeordnete Apps die gleichen Rollen haben können, handelt es sich im eine
|
||||
# "many-to-many"-Relation, die in dieser Hilfstabelle verwaltet werden
|
||||
userapps_roles = dbauth.Table(
|
||||
'userapps_roles',
|
||||
Column('userappid', Integer, ForeignKey('userapps.id'), primary_key=True),
|
||||
Column('roleid', Integer, ForeignKey('roles.id'), primary_key=True),
|
||||
bind_key=BINDKEY
|
||||
)
|
||||
|
||||
class Role(dbauth.Model):
|
||||
__tablename__ = 'roles'
|
||||
__bind_key__ = BINDKEY
|
||||
id = Column(Integer, primary_key=True)
|
||||
roleid = Column(String(80), index=True, unique=True, nullable=False, default=None)
|
||||
name = Column(String(80), nullable=False, default=None)
|
||||
level = Column(Integer, default=0)
|
||||
beschreibung = Column(String(800))
|
||||
|
||||
def to_dict(self):
|
||||
if self.level is None:
|
||||
self.level = 0
|
||||
return columns2dict(self)
|
||||
|
||||
# Die Tabelle "UserApp" enthält für jeden User für jede App, in der der User eine Rolle hat,
|
||||
# einen Datensatz.
|
||||
class Userapp(dbauth.Model):
|
||||
__tablename__ = 'userapps'
|
||||
__bind_key__ = BINDKEY
|
||||
id = Column(Integer, primary_key=True)
|
||||
userid = Column(ForeignKey('users.id'), index=True)
|
||||
appid = Column(ForeignKey('applications.id'))
|
||||
userapp_roles = dbauth.relationship('Role', secondary=userapps_roles, lazy='subquery', order_by="asc(Role.roleid)")
|
||||
application = dbauth.relationship('Application', uselist=False)
|
||||
|
||||
def get_role(self, roleid: str) -> Role:
|
||||
"""
|
||||
Die Methode gibt die Rolle aus der userapp zurück, die mit der roleid (nicht id) angegeben wurde.
|
||||
Gibt es keine Rolle mit der übergebenen roleid, wird 'None' zurückgegeben
|
||||
- roleid: str
|
||||
- die roleid (nicht id) der Rollue, die gesucht wird
|
||||
"""
|
||||
for role in self.userapp_roles:
|
||||
if role.id == roleid:
|
||||
return role
|
||||
return None
|
||||
|
||||
def to_dict(self, withrelations=True):
|
||||
result = columns2dict(self)
|
||||
result["userapp_roles"] = []
|
||||
if withrelations:
|
||||
for userapprole in self.userapp_roles:
|
||||
result["userapp_roles"].append(userapprole.to_dict())
|
||||
return result
|
||||
|
||||
class User(dbauth.Model):
|
||||
__tablename__ = 'users'
|
||||
__bind_key__ = BINDKEY
|
||||
id = Column(Integer, primary_key=True)
|
||||
userid = Column(String(80), index=True, unique=True, nullable=False, default='')
|
||||
username = Column(String(80), nullable=False)
|
||||
password = Column(String(40), nullable=False)
|
||||
email = Column(String(120))
|
||||
telefon = Column(String(800))
|
||||
# Jeder User kann mehrere "User-Apps" haben. Dies sind die Apps, in denen dem User Rollen
|
||||
# zugewiesen wurden. Es handelt sich um eine "one-to-many"-Relation, da ein User zwar mehrere
|
||||
# User-Apps haben kann, jede User-App aber genau einem User gehört
|
||||
userapps = dbauth.relationship('Userapp')
|
||||
|
||||
def get_userapp(self, appid:str) -> Userapp:
|
||||
"""
|
||||
Die Funktion liefert die userapp mit Hilfe der appid (nicht der id) zurück.
|
||||
Wenn der User keine userapp für die Application hat, wird 'None' zurückgegeben
|
||||
|
||||
- appid: string
|
||||
- Der Kurzname der Application (nicht die id)
|
||||
"""
|
||||
for userapp in self.userapps:
|
||||
app = userapp.application
|
||||
if app.appid == appid:
|
||||
return userapp
|
||||
return None
|
||||
|
||||
def get_userapp_by_id(self, id:int) -> Userapp:
|
||||
"""
|
||||
Die Funktion liefert die userapp mit Hilfe der id (nicht der appid) zurück
|
||||
|
||||
- appid: int
|
||||
- die interne Record-ID der Application (nicht die userid)
|
||||
"""
|
||||
for userapp in self.userapps:
|
||||
app = userapp.application
|
||||
if app.id == id:
|
||||
return userapp
|
||||
return None
|
||||
|
||||
def get_userapproles(self, appid:str) -> List[Role]:
|
||||
"""
|
||||
Die Funktion liefert die roles zurück, die der User in der durch die appid (nicht id) bestimmten Application hat
|
||||
|
||||
- appid: str
|
||||
- Der Kurzname der Application
|
||||
"""
|
||||
userapp: Userapp = self.get_userapp(appid)
|
||||
if userapp is not None:
|
||||
return userapp.userapp_roles
|
||||
else:
|
||||
return []
|
||||
|
||||
def has_roles(self, roleids: List[str]) -> bool:
|
||||
"""
|
||||
Die Funktion liefert 'True' zurück, falls der User in irgendeiner Application eine der Rollen hat, deren roleids
|
||||
übergeben wurden, ansonsten 'False'
|
||||
|
||||
- roleids: List[str]
|
||||
- eine Liste mit roleids (nicht ids)
|
||||
"""
|
||||
for userapp in self.userapps:
|
||||
for role in userapp.userapp_roles:
|
||||
if role.roleid in roleids:
|
||||
return True
|
||||
return False
|
||||
|
||||
def is_SysAdmin(self) -> bool:
|
||||
"""
|
||||
die Methode liefert 'True' zurück, wenn der User in irgendeiner Anwendung SysAdmin ist
|
||||
"""
|
||||
return self.has_roles([SYS_ADMIN])
|
||||
|
||||
def __repr__(self):
|
||||
return '<User %r>' % self.username
|
||||
|
||||
def to_dict(self, withrelations=False):
|
||||
result = columns2dict(self, excludefields=['password'])
|
||||
result['userapps'] = []
|
||||
if withrelations:
|
||||
for userapp in self.userapps:
|
||||
result['userapps'].append(userapp.to_dict())
|
||||
return result
|
||||
|
||||
def to_relationaldict(self):
|
||||
return self.to_dict(withrelations=True)
|
||||
|
||||
def has_permission(self, appid:str, allowed_roles: List[str]):
|
||||
"""
|
||||
Gibt an, ob der User in einer bestimmten Anwendung eine der gewünschten Rollen hat
|
||||
- appid: string
|
||||
- Die appid (nicht id) der gewünschten Application
|
||||
- allowed_roles: List[str]
|
||||
- Eine Liste mit den roleids (nicht ids) der erlaubten Rollen
|
||||
"""
|
||||
userapp = self.get_userapp(appid)
|
||||
if userapp is not None:
|
||||
rolleerlaubt = any([role.roleid in allowed_roles for role in userapp.userapp_roles])
|
||||
if rolleerlaubt:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
|
||||
# Die nachfolgenden Properties werden für flask_login benötigt
|
||||
@property
|
||||
def is_authenticated(self):
|
||||
return True
|
||||
|
||||
@property
|
||||
def is_active(self):
|
||||
return True
|
||||
|
||||
@property
|
||||
def is_anonymous(self):
|
||||
return False
|
||||
|
||||
def get_id(self):
|
||||
return self.userid
|
||||
|
||||
class Application(dbauth.Model):
|
||||
__tablename__ = 'applications'
|
||||
__bind_key__ = BINDKEY
|
||||
id = Column(Integer, primary_key=True)
|
||||
appid = Column(String(80), index=True, unique=True, nullable=False, default=None)
|
||||
name = Column(String(80), nullable=False, default=None)
|
||||
minversion = Column(String(20))
|
||||
beschreibung = Column(String(800))
|
||||
roles = dbauth.relationship('Role', secondary=apps_roles, lazy='subquery', order_by="asc(Role.roleid)")
|
||||
|
||||
def to_dict(self, withrelations=True):
|
||||
result = columns2dict(self)
|
||||
result['roles'] = []
|
||||
for role in self.roles:
|
||||
result['roles'].append(role.to_dict())
|
||||
return result
|
||||
|
||||
print("########## auth/model importiert")
|
||||
@@ -0,0 +1,73 @@
|
||||
from . import APP_ADMIN, SYS_ADMIN, User, Userapp, dbauth, Application, Role, apps_roles, userapps_roles
|
||||
from .errors import AutorisationError, ApplicationError
|
||||
from flask_login import current_user
|
||||
from sqlalchemy import delete
|
||||
|
||||
session = dbauth.session
|
||||
|
||||
def get_applications():
|
||||
apps = []
|
||||
for app in Application.query:
|
||||
# Die Application wird nur zurückgegeben, wenn der current-User SysAdmin ist oder wenn
|
||||
# der current_user in der zurückzugebenden Application AppAdmin ist
|
||||
if current_user.is_SysAdmin() or current_user.has_permission(app.appid, [APP_ADMIN]):
|
||||
if not current_user.is_SysAdmin():
|
||||
for role in list(app.roles):
|
||||
if role.roleid == SYS_ADMIN:
|
||||
app.roles.remove(role)
|
||||
apps.append(app)
|
||||
return apps
|
||||
|
||||
def updateapplication(data):
|
||||
if not current_user.is_SysAdmin():
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN], user_roles=current_user.get_userapproles(data["appid"]))
|
||||
currentapp = None
|
||||
if data["id"] == -1:
|
||||
currentapp = Application()
|
||||
currentapp.id = None
|
||||
session.add(currentapp)
|
||||
else:
|
||||
currentapp = Application.query.get(data["id"])
|
||||
if currentapp is None:
|
||||
raise Exception ("404")
|
||||
currentapp.name = data["name"]
|
||||
currentapp.appid = data["appid"]
|
||||
currentapp.minversion = data["minversion"]
|
||||
currentapp.beschreibung = data["beschreibung"]
|
||||
session.commit()
|
||||
|
||||
def deleteapplications(appids):
|
||||
if not current_user.is_SysAdmin():
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN], user_roles=None)
|
||||
for appid in appids:
|
||||
app: Application = Application.query.filter(Application.appid == appid).one()
|
||||
if app is not None:
|
||||
# Die Datensätze in "apps_roles" müssen gelöscht werden
|
||||
app.roles.clear()
|
||||
# UserApps löschen
|
||||
q_userapps = dbauth.select(Userapp). where(Userapp.appid == app.id)
|
||||
for userapp in dbauth.session.scalars(q_userapps):
|
||||
userapp.userapp_roles.clear()
|
||||
session.delete(userapp)
|
||||
session.delete(app)
|
||||
session.commit()
|
||||
|
||||
def updateapplicationroles(approles):
|
||||
# Änderungen an Applications darf grundsätzlich nur der SysAdmin machen
|
||||
if not current_user.is_SysAdmin():
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN], user_roles=None)
|
||||
|
||||
for appidinfo in approles:
|
||||
app: Application = Application.query.filter(Application.appid == appidinfo["appid"]).one()
|
||||
if app is None:
|
||||
raise ApplicationError(f"Die Application mit der Appid {appidinfo['appid']} existiert nicht", )
|
||||
for roleidinfo in appidinfo["roles"]:
|
||||
role: Role = Role.query.filter(Role.roleid == roleidinfo["roleid"]).one()
|
||||
if role in app.roles:
|
||||
if not roleidinfo['isChecked']:
|
||||
app.roles.remove(role) # hier kann das remove einfach so verwendet werden, da nicht über app.roles iteriert wird
|
||||
# TODO: die Rolle muss auch aus allen userapps für alle User gelöscht werden
|
||||
else:
|
||||
if roleidinfo['isChecked']:
|
||||
app.roles.append(role)
|
||||
session.commit()
|
||||
@@ -0,0 +1,86 @@
|
||||
from datetime import datetime
|
||||
from auth import model
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.exc import OperationalError
|
||||
from datetime import datetime
|
||||
from db import dbauth
|
||||
|
||||
session = dbauth.session
|
||||
BINDKEY = "auth"
|
||||
|
||||
def _db_V01_00():
|
||||
# "bind_key=None" bedeutet hier, dass bei "create_all" nur die Models mit
|
||||
# Default-Bind erzeugt werden sollen und nicht die anderen, insbesonder
|
||||
# nicht die zum MS-SQL-SErver
|
||||
dbauth.create_all(bind_key=BINDKEY)
|
||||
systeminfo = model.Systeminfo()
|
||||
systeminfo.id = 1
|
||||
systeminfo.version = "01.00"
|
||||
systeminfo.lastupdate = datetime.now()
|
||||
systeminfo.initialinstall = datetime.now()
|
||||
session.add(systeminfo)
|
||||
session.commit()
|
||||
|
||||
def _db_V01_01():
|
||||
u = model.User()
|
||||
u.id = 1
|
||||
u.userid = 'System.Admin'
|
||||
u.username = 'Systemadministrator'
|
||||
u.password = "geheim"
|
||||
u.email = ''
|
||||
session.add(u)
|
||||
|
||||
sysmgr = model.Role()
|
||||
sysmgr.id = 1
|
||||
sysmgr.name = "Systemadministrator"
|
||||
sysmgr.roleid = "SysAdmin"
|
||||
session.add(sysmgr)
|
||||
|
||||
appmgr = model.Role()
|
||||
appmgr.name = "Anwendungsadministrator"
|
||||
appmgr.roleid = "AppAdmin"
|
||||
session.add(appmgr)
|
||||
|
||||
usermgr = model.Role()
|
||||
usermgr.name = "Benutzerverwaltung"
|
||||
usermgr.roleid = "UserMgr"
|
||||
session.add(usermgr)
|
||||
|
||||
application = model.Application()
|
||||
application.id = 1
|
||||
application.appid = "Project01"
|
||||
application.name = "Project01 (Name ändern)"
|
||||
session.add(application)
|
||||
|
||||
userapp = model.Userapp()
|
||||
userapp.application = application
|
||||
userapp.userapp_roles.append(sysmgr)
|
||||
session.add(userapp)
|
||||
u.userapps.append(userapp)
|
||||
|
||||
_updatesysteminfo('01.01')
|
||||
session.commit()
|
||||
|
||||
def _updatesysteminfo(versionnr):
|
||||
s: model.Systeminfo = model.Systeminfo.query.get(1)
|
||||
s.version = versionnr
|
||||
s.lastupdate = datetime.now()
|
||||
return
|
||||
|
||||
def checkupdates():
|
||||
systeminfo = None
|
||||
curr_dbversion = '_db_V01_00'
|
||||
try:
|
||||
systeminfo: model.Systeminfo = model.Systeminfo.query.get(1)
|
||||
curr_dbversion = "_db_V" + systeminfo.version.replace(".", "_")
|
||||
except (OperationalError, AttributeError):
|
||||
_db_V01_00()
|
||||
|
||||
versionen = sorted(filter(lambda f: f.startswith("_db_V"),globals()))
|
||||
|
||||
for v in versionen:
|
||||
if v > curr_dbversion:
|
||||
globals()[v]() # https://docs.python.org/3.6/library/functions.html?highlight=globals#globals
|
||||
|
||||
print('########## Database updated')
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
|
||||
class ApplicationError(Exception):
|
||||
def __init__(self, *messages, htmlstatus=400):
|
||||
self.htmlstatus = htmlstatus
|
||||
Exception.__init__(self, *messages)
|
||||
|
||||
@property
|
||||
def error(self):
|
||||
error = {
|
||||
"status": self.htmlstatus,
|
||||
"messages": list(self.args)
|
||||
}
|
||||
return error
|
||||
|
||||
class AutorisationError(ApplicationError):
|
||||
def __init__(self, allowed_roles=[], user_roles=[]):
|
||||
self.allowed_roles = allowed_roles
|
||||
self.user_roles = user_roles
|
||||
|
||||
if user_roles is None:
|
||||
userrolestxt = "NoSysAdmin"
|
||||
else:
|
||||
userrolestxt = ", ".join([role.name for role in self.user_roles])
|
||||
|
||||
ApplicationError.__init__(self,
|
||||
"Sie haben nicht die passende Rolle für diese Funktion",
|
||||
"Ihre Rollen sind: " + userrolestxt,
|
||||
"Erforderliche Rollen sind: " + ", ".join(self.allowed_roles), htmlstatus=HtmlStatusCode.Unauthorized)
|
||||
|
||||
class HtmlStatusCode():
|
||||
Bad_Request = 400
|
||||
Unauthorized = 401
|
||||
Payment_Required = 402
|
||||
Forbidden = 403
|
||||
Not_Found = 404
|
||||
Method_Not_Allowed = 405
|
||||
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
from . import SYS_ADMIN, Application, Role, dbauth, User
|
||||
from sqlalchemy.sql import collate
|
||||
from flask_login import current_user
|
||||
from .errors import AutorisationError
|
||||
|
||||
session = dbauth.session
|
||||
|
||||
def get_roles():
|
||||
roles = Role.query.order_by(collate(Role.roleid, 'NOCASE')).all()
|
||||
return roles
|
||||
|
||||
def updaterole(data):
|
||||
if not current_user.is_SysAdmin():
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN], user_roles=None)
|
||||
currentrole = None
|
||||
if data["id"] == -1:
|
||||
currentrole = Role()
|
||||
currentrole.id = None
|
||||
session.add(currentrole)
|
||||
else:
|
||||
currentrole = Role.query.get(data["id"])
|
||||
if currentrole is None:
|
||||
raise Exception ("404")
|
||||
currentrole.level = data["level"]
|
||||
currentrole.name = data["name"]
|
||||
currentrole.roleid = data["roleid"]
|
||||
currentrole.beschreibung = data["beschreibung"]
|
||||
session.commit()
|
||||
|
||||
def deleteroles(roleids):
|
||||
if not current_user.is_SysAdmin():
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN], user_roles=None)
|
||||
for roleid in roleids:
|
||||
role: Role = Role.query.filter(Role.roleid == roleid).one()
|
||||
if role is not None:
|
||||
appq = Application.query
|
||||
for app in appq:
|
||||
if role in app.roles:
|
||||
app.roles.remove(role) # remove kann hier einfach so verwendet werden, da nicht über app.roles iteriert wird
|
||||
userq = User.query
|
||||
for user in userq:
|
||||
for userapp in user.userapps:
|
||||
if role in userapp.userapp_roles:
|
||||
userapp.userapp_roles.remove(role) # remove kann hier einfach so verwendet werden, da nicht über userapp_roles iteriert wird
|
||||
session.delete(role)
|
||||
|
||||
session.commit()
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
|
||||
from . import APP_ADMIN, SYS_ADMIN, Application, Role, User, dbauth, Userapp
|
||||
from .errors import ApplicationError, AutorisationError
|
||||
from flask_login import current_user
|
||||
current_user: User
|
||||
|
||||
session = dbauth.session
|
||||
|
||||
def check_login(userid: str, password: str):
|
||||
"""
|
||||
Hier wird ausschließlich nur festgestellt, ob Login und Kennwort vorhanden sind bzw. übereinstimmen.
|
||||
alles Weitere muss in den Funktionen erfolgen, die 'check_login' aufrufen.
|
||||
"""
|
||||
user:User = User.query.filter(User.userid == userid).first()
|
||||
if user is None or user.password != password:
|
||||
return None
|
||||
else:
|
||||
return user
|
||||
|
||||
def get_users():
|
||||
"""
|
||||
Hier werden derzeit alle User zurückgegeben.
|
||||
Allerdings werden die userapps eines Users wie folgt gefiltert:
|
||||
|
||||
Wenn es sich bei dem current_user um den SysAdmin handelt, wird nichts gefiltert
|
||||
|
||||
Wenn es sich bei dem current_user um den AppAdmin handelt, werden nur die userapps zurückgegeben, auf die auch
|
||||
der current_user Zugriff hat. Außerdem wird die Rolle "SysAdmin" innerhalb der userapp.userapp-roles auch nicht
|
||||
mit zurückgegeben
|
||||
|
||||
Bei allen anderen Arten von Usern werden keine userapps und somit auch keine Rollen zurückgegeben
|
||||
"""
|
||||
users = []
|
||||
user: User
|
||||
for user in User.query.order_by(User.userid):
|
||||
# Wenn der current_user kein SysAdmin ist, soll diese User auch nicht sehen, dass es einen SysAdmin gibt
|
||||
if not current_user.is_SysAdmin() and user.is_SysAdmin():
|
||||
continue
|
||||
# Wenn innerhalb einer For-Schleife Elemente gelöscht werden, muss auf einer Kopie der
|
||||
# Liste iteriert werden, da die For-Schleife sonst mit der Nummerierung durcheinanderkommt, weil während
|
||||
# des Durchlaufs Werte durch das remove "verschwinden"
|
||||
# Daher wird hier für die For-Schleife mit der Funktion list() eine neue Liste der Origonal-Liste erzeugt
|
||||
for userapp in list(user.userapps):
|
||||
# Rollen für bestimmte Application, in denen der zurückzugebende User Rollen innehat, sollen nur zurück-
|
||||
# gegeben werden, wenn der current_user für diese App SysAdmin oder AppAdmin ist
|
||||
if not current_user.has_permission(userapp.application.appid, [APP_ADMIN, SYS_ADMIN]):
|
||||
user.userapps.remove(userapp)
|
||||
# Wenn der current_user für die Application keine SysAdmin-Rolle hat, soll eine evtl. SysAdmin-Rolle auch
|
||||
# nicht zurückgegeben werden
|
||||
if not current_user.has_permission(userapp.application.appid, [SYS_ADMIN]):
|
||||
for role in list(userapp.userapp_roles): # s.o. wegen remove in for-loop
|
||||
if role.roleid == SYS_ADMIN:
|
||||
userapp.userapp_roles.remove(role)
|
||||
users.append(user)
|
||||
return users
|
||||
|
||||
|
||||
def changepassword(userid:str, oldpassword:str, newpassword1:str, newpassword2:str):
|
||||
"""
|
||||
Die Funktion ändert das Kennwort des Users
|
||||
|
||||
-userid: string Die Userid (nicht id) des Users
|
||||
-oldpassword: string Das bisherige Kennwort des Users
|
||||
-newpassword1: string Das neue Kennwort des Users
|
||||
-newpassword2: string Die Wiederholung des neuen Kennwort des Users
|
||||
"""
|
||||
if newpassword1 != newpassword2:
|
||||
raise ValueError("Die neuen Kennworte stimmen nicht überein!")
|
||||
if newpassword1 is None or newpassword1.strip() == "":
|
||||
raise ValueError("Es muss ein neues Kennwort angegeben werden!")
|
||||
user:User = User.query.filter(User.userid == userid).first()
|
||||
if user is None or user.password != oldpassword:
|
||||
raise ValueError("Der angegebene Benutzer existiert nicht oder das alte Kennwort ist falsch")
|
||||
user.password = newpassword1
|
||||
session.commit()
|
||||
return True
|
||||
|
||||
def updateuser(data):
|
||||
"""
|
||||
Die User-Daten werden nur geändert, wenn der current_user entweder SysAdmin oder AppAdmin ist
|
||||
-data: Dict Ein Dictionary mit den Feldern, die beim User geändert werden sollen
|
||||
"""
|
||||
if not current_user.has_roles([SYS_ADMIN, APP_ADMIN]):
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN, APP_ADMIN], user_roles=None)
|
||||
currentuser: User = None
|
||||
if data["id"] == -1:
|
||||
currentuser = User()
|
||||
currentuser.id = None
|
||||
currentuser.password = ""
|
||||
session.add(currentuser)
|
||||
else:
|
||||
currentuser = User.query.get(data["id"])
|
||||
if currentuser is None:
|
||||
raise Exception ("404")
|
||||
currentuser.username = data["username"]
|
||||
currentuser.userid = data["userid"]
|
||||
currentuser.email = data["email"]
|
||||
currentuser.telefon = data["telefon"]
|
||||
session.commit()
|
||||
|
||||
def deleteusers(userids):
|
||||
"""
|
||||
Hier werden die User gelöscht, deren userids übergeben werden. Gelöscht werden kann aber nur,
|
||||
wenn der current_user entweder SysAdmin oder AppAdmin ist.
|
||||
Wenn der zu löschende User Rollen in Applications hat, auf die der AppAdmin keinen Zugriff hat,
|
||||
Wird der User ebenfalls nicht gelöscht.
|
||||
- userids: string[] eine Liste der userids (nicht id) der User, die gelöscht werden sollen
|
||||
"""
|
||||
# Wenn er schon mal nirgendwo Sys- oder AppAdmin ist, fliegt er schon mal direkt raus
|
||||
if not current_user.has_roles([SYS_ADMIN, APP_ADMIN]):
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN, APP_ADMIN], user_roles=None)
|
||||
# ansonsten: hier dürfen nur die User gelöscht werden, die in denjenigen Apps eine Rolle haben,
|
||||
# in dene der current_user entweder SysAdmin aod AppAdmin ist. Wenn der User noch eine Rolle in einer App haben,
|
||||
# auf die der current_user keinen Zugriff hat, darf nicht gelöscht werden
|
||||
for userid in userids:
|
||||
user: User = User.query.filter(User.userid == userid).one()
|
||||
if user is not None:
|
||||
for userapp in user.userapps:
|
||||
currentuser_userapp: Userapp = current_user.get_userapp(userapp.application.appid)
|
||||
if currentuser_userapp is None or len(currentuser_userapp.userapp_roles) == 0:
|
||||
raise ApplicationError(f"Der Benutzer {user.username} hat noch Rollen in Anwendungen, in denen Sie kein AppAdmin oder SysAdmin sind.",
|
||||
"Er kann daher nicht gelöscht werden.",
|
||||
"Der komplette Löschvorgang wurde abgebrochen.",
|
||||
htmlstatus=404)
|
||||
session.delete(user)
|
||||
session.commit()
|
||||
|
||||
def applyuserapprole(data):
|
||||
"""
|
||||
Hier kann einem Benutzer eine Rolle für eine bestimmte App entzogen oder zugewiesen werden
|
||||
|
||||
### Parameter
|
||||
- data: dict
|
||||
- parentuserid: integer
|
||||
- Die id (nicht userid) des Users, für den eine Rolle geändert werden soll
|
||||
- parantappid: integer
|
||||
- Die id (nicht appid) der Application, für die eine Rolle geändert werden soll
|
||||
- roleid: integer
|
||||
- Die id (nicht roleid) der Rolle, die zugewiesen oder entzogen werden soll
|
||||
- ischecked: boolean
|
||||
- 'True' bedeutet, dass die Rolle zugewiesen werden soll, 'False' bedeutet, dass die Rolle entzogen werden soll
|
||||
"""
|
||||
user_id = data['parentuserid']
|
||||
app_id = data['parentappid']
|
||||
ischecked = data['ischecked']
|
||||
role_id = data['roleid']
|
||||
|
||||
user: User = User.query.get(user_id)
|
||||
if user is None:
|
||||
raise Exception("Den User gibt es nicht")
|
||||
app: Application = Application.query.get(app_id)
|
||||
if app is None:
|
||||
raise Exception("Die Application gibt es nicht")
|
||||
role: Role = Role.query.get(role_id)
|
||||
if role is None:
|
||||
raise Exception("Die Rolle gibt es nicht")
|
||||
|
||||
# hier darf nur zugewiesen werden, wenn der current_user entweder SysAdmin ist oder AppAdmin in der App,
|
||||
# für die die Rolle geändert wird
|
||||
if not current_user.is_SysAdmin() and not current_user.has_permission(app.appid, [APP_ADMIN]):
|
||||
raise AutorisationError(allowed_roles=[SYS_ADMIN, APP_ADMIN], user_roles=current_user.get_userapproles(app.appid))
|
||||
userapp: Userapp = user.get_userapp_by_id(app_id)
|
||||
if userapp is None:
|
||||
userapp = Userapp()
|
||||
userapp.appid = app_id
|
||||
user.userapps.append(userapp)
|
||||
userapp_role: Role = userapp.get_role(role_id)
|
||||
if userapp_role is None and ischecked:
|
||||
userapp.userapp_roles.append(role)
|
||||
if userapp_role is not None and not ischecked:
|
||||
# eine SysAdmin oder ein AppAdmin kann sich selber nicht die Rolle als SysAdmin oder AppAdmin entziehen
|
||||
if current_user.userid == user.userid:
|
||||
if ((userapp_role.roleid == APP_ADMIN and current_user.has_permission(app.appid, [APP_ADMIN])) or
|
||||
(userapp_role.roleid == SYS_ADMIN and current_user.has_permission(app.appid, [SYS_ADMIN]))):
|
||||
raise ApplicationError("Sie können sich nicht selbst die Rolle als AppAdmin entziehen, da Sie dann keinen Zugriff auf das System mehr hätten!")
|
||||
userapp.userapp_roles.remove(userapp_role) # remove kann hier einfach so verwendet werden, da nicht über userapp.userapp_roles iteriert wird
|
||||
session.commit()
|
||||
|
||||
|
||||
@@ -0,0 +1,235 @@
|
||||
from approot import login_manager
|
||||
from auth.model import (
|
||||
APP_ADMIN,
|
||||
APP_DOZENT,
|
||||
APP_USER,
|
||||
SYS_ADMIN,
|
||||
Application,
|
||||
Role,
|
||||
User,
|
||||
applications,
|
||||
roles,
|
||||
users,
|
||||
)
|
||||
from flask import current_app as app
|
||||
from flask import jsonify, request
|
||||
from flask_login import login_required, login_user, logout_user
|
||||
|
||||
from .model.errors import ApplicationError, AutorisationError
|
||||
|
||||
print("auth-routes")
|
||||
|
||||
|
||||
@login_manager.user_loader
|
||||
def user_loader(userid):
|
||||
user: User | None = User.query.filter(User.userid == userid).first()
|
||||
return user
|
||||
|
||||
|
||||
@app.post("/auth/checklogin")
|
||||
def checklogin():
|
||||
"""
|
||||
Hier wird nur die Authentifizierung überprüft.
|
||||
Evtl. Autorisierungen anhand der User-Rollen und App-Berechtigungen müssen später bei den
|
||||
Funktionsaufrufen erfolgen
|
||||
"""
|
||||
data = request.json
|
||||
user: User = users.check_login(data["userid"], data["password"])
|
||||
if user is None:
|
||||
error = ApplicationError(
|
||||
"Login oder Kennwort stimmen nicht!",
|
||||
"Anmeldung wurde verweigert",
|
||||
htmlstatus=403,
|
||||
)
|
||||
return jsonify(error.error), error.htmlstatus
|
||||
if not user.has_roles([SYS_ADMIN, APP_ADMIN, APP_USER, APP_DOZENT]):
|
||||
error = ApplicationError(
|
||||
"Sie haben nicht die passende Rolle zur Nutzung dieser Anwendung.",
|
||||
"Der Zugang wird verweigert",
|
||||
htmlstatus=403,
|
||||
)
|
||||
return jsonify(error.error), error.htmlstatus
|
||||
login_user(user)
|
||||
result = user.to_dict()
|
||||
result["isSysAdmin"] = user.is_SysAdmin()
|
||||
return jsonify(result), 200
|
||||
|
||||
|
||||
@app.post("/auth/changepassword")
|
||||
@login_required
|
||||
def changepassword():
|
||||
data = request.json
|
||||
status = 200
|
||||
result = {"Message": "", "Success": True}
|
||||
try:
|
||||
users.changepassword(
|
||||
data["Userid"],
|
||||
data["OldPassword"],
|
||||
data["NewPassword1"],
|
||||
data["NewPassword2"],
|
||||
)
|
||||
except Exception as ex:
|
||||
error = ApplicationError(str(ex), htmlstatus=400)
|
||||
return error.error, error.htmlstatus
|
||||
|
||||
return jsonify(result), status
|
||||
|
||||
|
||||
@app.get("/auth/logout")
|
||||
def logout():
|
||||
logout_user()
|
||||
return jsonify({"logout": True}), 200
|
||||
|
||||
|
||||
@app.get("/auth/getusers")
|
||||
@login_required
|
||||
def getusers():
|
||||
"""
|
||||
Die Rückgabe wird in 'users.get_users()' (s.u.) anhand er App-Berechtigung und der
|
||||
Rollen gefiltert.
|
||||
"""
|
||||
userquery = users.get_users()
|
||||
result = []
|
||||
user: User
|
||||
for user in userquery:
|
||||
result.append(user.to_relationaldict())
|
||||
return jsonify(result), 200
|
||||
|
||||
|
||||
@app.get("/auth/getapplications")
|
||||
@login_required
|
||||
def getapplications():
|
||||
"""
|
||||
Die Rückgabe wird in 'applications.get_applications()' (s.u.) anhand er App-Berechtigung und der
|
||||
Rollen gefiltert.
|
||||
"""
|
||||
appquery = applications.get_applications()
|
||||
result = []
|
||||
app: Application
|
||||
for app in appquery:
|
||||
result.append(app.to_dict())
|
||||
return jsonify(result), 200
|
||||
|
||||
|
||||
@app.get("/auth/getroles")
|
||||
@login_required
|
||||
def getroles():
|
||||
rolelist = roles.get_roles()
|
||||
result = []
|
||||
role: Role
|
||||
for role in rolelist:
|
||||
result.append(role.to_dict())
|
||||
return jsonify(result), 200
|
||||
|
||||
|
||||
@app.post("/auth/updaterole")
|
||||
@login_required
|
||||
def updaterole():
|
||||
data = request.json
|
||||
try:
|
||||
roles.updaterole(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
if str(ex) == "404":
|
||||
return jsonify("Rolle nicht gefunden"), 404
|
||||
else:
|
||||
return jsonify(str(ex)), 400
|
||||
return jsonify({}), 200
|
||||
|
||||
|
||||
@app.post("/auth/deleteroles")
|
||||
@login_required
|
||||
def deleteroles():
|
||||
data = request.json
|
||||
try:
|
||||
roles.deleteroles(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
return jsonify(str(ex)), 400
|
||||
return jsonify({}), 200
|
||||
|
||||
|
||||
@app.post("/auth/updateapplication")
|
||||
@login_required
|
||||
def updateapplication():
|
||||
data = request.json
|
||||
try:
|
||||
applications.updateapplication(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
return jsonify(str(ex)), 400
|
||||
return jsonify({}), 200
|
||||
|
||||
|
||||
@app.post("/auth/deleteapplications")
|
||||
@login_required
|
||||
def deleteapplications():
|
||||
data = request.json
|
||||
try:
|
||||
applications.deleteapplications(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
return str(ex), 400
|
||||
return {}, 200
|
||||
|
||||
|
||||
@app.post("/auth/updateapplicationroles")
|
||||
@login_required
|
||||
def updateapplicationroles():
|
||||
data = request.json
|
||||
try:
|
||||
applications.updateapplicationroles(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
return jsonify(str(ex)), 400
|
||||
return jsonify({}), 200
|
||||
|
||||
|
||||
@app.post("/auth/updateuser")
|
||||
@login_required
|
||||
def updateuser():
|
||||
data = request.json
|
||||
try:
|
||||
users.updateuser(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
if str(ex) == "404":
|
||||
return jsonify("Benutzer nicht gefunden"), 404
|
||||
else:
|
||||
return jsonify(str(ex)), 400
|
||||
return jsonify({}), 200
|
||||
|
||||
|
||||
@app.post("/auth/applyuserapprole")
|
||||
@login_required
|
||||
def applyuserapprole():
|
||||
data = request.json
|
||||
try:
|
||||
users.applyuserapprole(data)
|
||||
except AutorisationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
return jsonify(str(ex)), 400
|
||||
return jsonify({}), 200
|
||||
|
||||
|
||||
@app.post("/auth/deleteusers")
|
||||
@login_required
|
||||
def deleteusers():
|
||||
data = request.json
|
||||
try:
|
||||
users.deleteusers(data)
|
||||
except ApplicationError as ex:
|
||||
return jsonify(ex.error), ex.htmlstatus
|
||||
except Exception as ex:
|
||||
return jsonify(str(ex)), 400
|
||||
return {}, 200
|
||||
|
||||
|
||||
print("########## auth.routes importiert")
|
||||
Reference in new issue
Block a user