# (c) cavaliba.com - IAM - aaa.py import base64 import re from pprint import pprint import app_home.cache as cache from app_home.configuration import get_configuration from app_home.log import DEBUG, ERROR, INFO, WARNING, log from app_user.permission import permission_all_keynames #from app_user.role import role_get_by_name from app_user.role import role_get_permissions from app_user.user import user_create, user_get_by_id, user_get_by_login from django.conf import settings from django.forms.models import model_to_dict from django.utils import timezone from .ip import get_user_ip, is_trusted_ip from .models import SireneGroup, SireneUser, SireneVisitor # ============================================================================== # update_last_login # ============================================================================== def update_last_login(aaa): if aaa["userid"]: user = user_get_by_id(aaa["userid"]) if user: user.last_login = timezone.now() try: user.save(update_fields=["last_login"]) except Exception: pass elif aaa["is_visitor"]: username = aaa["username"] if not username: return if len(username) == 0: return visitor = SireneVisitor.objects.filter(username=username).first() if not visitor: visitor = SireneVisitor(username=username) visitor.last_login = timezone.now() visitor.user_ip = aaa.get("user_ip", "") visitor.save() return # ============================================================================== # start_command # ============================================================================== def start_command(command="n/a", action="n/a"): cache.init() log(WARNING, app="command", view=command, action=action, data="start") return # ============================================================================== # start_ajax # ============================================================================== def start_ajax(request): cache.init() if get_configuration(appname="env", keyname="CAVALIBA_DEBUG_AAA") == "yes": aaa = get_aaa(request) request.session["aaa"] = aaa pprint(aaa) else: # session ? if 'aaa' in request.session: aaa = request.session['aaa'] else: aaa = get_aaa(request) request.session["aaa"] = aaa cache.cache_aaa = aaa context = {} context["aaa"] = aaa return context # ============================================================================== # start_view # - load conf in cache (refresh for each view) # - check existing session => get login / age # - get_aaa => redirect if KO # - check perm for view => redirect if KO # - log # ============================================================================== def start_view(request, app="na", view="na", noauth=None, perm=None, noauthz=None, visitor=False): ''' returns a context w/ a redirect target if auth or autor i''' cache.init() # visitor access need explicit visitor=True debug_aaa = get_configuration(appname="env", keyname="CAVALIBA_DEBUG_AAA") cache_session = get_configuration(appname="user", keyname="CACHE_SESSION") if debug_aaa == "yes": print("------------------------------------") print(f"START VIEW - start for app={app} - view={view}") # already authenticated ? # get_aaa() or session cache if cache_session == "no": aaa = get_aaa(request) if debug_aaa == "yes": print("START VIEW - cache_session=no ; get_aaa() called") else: # session cache_hit ? if 'aaa' in request.session: aaa = request.session['aaa'] if debug_aaa == "yes": print("START VIEW - cache_session=yes ; cache_hit ; aaa in request.session") # session cache_miss else: aaa = get_aaa(request) if debug_aaa == "yes": print("START VIEW - cache_session=yes ; cache_miss ; get_aaa()") # only cache if successfully authenticated if aaa["is_authenticated"]: request.session["aaa"] = aaa if debug_aaa == "yes": print("START VIEW - cache miss + authenticated=yes ; set request.session") else: request.session.flush() if debug_aaa == "yes": print("START VIEW - cache_miss + authenticated = no ; flush session") cache.cache_aaa = aaa if debug_aaa == "yes": print("START VIEW get aaa result:") print(" auth_mode:", aaa["auth_mode"]) print(" impersonate_by:", aaa["impersonate_by"]) print(" is_anonymous:", aaa["is_anonymous"]) print(" is_authenticated:", aaa["is_authenticated"]) print(" is_admin:", aaa["is_admin"] ) print(" is_trusted_ip:", aaa["is_trusted_ip"]) print(" is_visitor:", aaa["is_visitor"]) print(" user_ip:", aaa["user_ip"]) print(" user_id:", aaa["userid"]) print(" username:", aaa["username"]) pprint(aaa["user"]) print(" --------- aaa:") pprint(aaa) print(" =========") context = {} context["aaa"] = aaa context["redirect"] = None update_last_login(aaa) # if param noauth provided; restricted access if noauth: # Django Internal Auth mode if aaa["auth_mode"] == "local": if not (aaa['is_authenticated'] or (visitor and aaa["is_visitor"])): context["redirect"] = f"{settings.LOGIN_URL}?next={request.path}" if debug_aaa == "yes": print("START VIEW : auth required, but local mode not authenticated") return context # TODO: direct ldap # elif aaa["auth_mode"] == "ldap": # if not (aaa['is_authenticated'] or (visitor and aaa["is_visitor"])): # context["redirect"]= "TO PUBLIC PAGE WITH LDAP LOGIN FORM" # return context # other (external modes) : federated, proxy-header else: if not (aaa['is_authenticated'] or (visitor and aaa["is_visitor"])): if debug_aaa == "yes": print("START VIEW : auth required, but not authenticated (other mode)") context["redirect"] = noauth return context if debug_aaa == "yes": print("START VIEW : check perm ; perm=",perm, aaa['perms']) if perm: if perm not in aaa["perms"]: log(WARNING, aaa=aaa, app=app, view=view, action="access", data="Not allowed") context["redirect"] = noauthz return context log(DEBUG, aaa=aaa, app=app, view=view, action="start_view", status="OK") return context # ---------------------------------------------------------------------- # get_aaa # ---------------------------------------------------------------------- # # auth_mode : oauth2, basic, local, forced, unittest (hidden) # # is_trusted_ip : user IP belongs to configured trusted list # is_anonymous : default - no session, no user info (no auth performed) # is_visitor : externally authenticated but not in SireneUser DB (e.g. external oauth) # is_authenticated : authenticated & exist in SireneUser DB (or admin) # - auth basic (HTTP header) # - fed/oauth2 # - sirene internal auth forms [TBD] # - session from previous # is_admin : built-in admin account # user : dict serialized SireneUser # username : login # userid : pk from DB # user_ip : # perms:[] : [perm_keyname, ...] # groups:[] : [groupnames, ...] => all groups for aaa # groups_direct:[] : [groupnames, ...] => direct membership (user is in group) # groups_computed:[] : [groupnames, ...] => computed (autogroup) groupnames for user # groups_indirect:[] : [groupnames, ...] => parent groups for direct_group # impersonate_by : login of source account ; default "" # ---------------------------------------------------------------------- def get_aaa(request, auth_mode = None, impersonate=None, impersonate_by=None): debug_aaa = get_configuration(appname="user", keyname="DEBUG_AAA") if len(cache.cache_aaa)>0: if debug_aaa == "yes": print("DEBUG AAA - get_aaa() from cache.cache_aaa.") return cache.cache_aaa aaa={} uname = "" email = "" aaa["is_trusted_ip"] = False aaa["is_anonymous"] = True aaa["is_authenticated"] = False aaa["is_visitor"] = False aaa["is_admin"] = False aaa['username'] = "unknown" aaa["user"] = None aaa["userid"] = None aaa["groups"] = [] aaa["groups_direct"] = [] aaa["groups_indirect"] = [] aaa["perms"] = [] aaa['user_ip'] = get_user_ip(request) aaa["is_trusted_ip"] = is_trusted_ip(aaa['user_ip']) aaa["impersonate_by"] = "" # ---------------- # get auth_mode # ---------------- if not auth_mode: auth_mode = get_configuration(appname="env", keyname="CAVALIBA_AUTH_MODE") aaa["auth_mode"] = auth_mode if debug_aaa == "yes": print("DEBUG_AAA - auth_mode: ", auth_mode) # ---------------- # get uname # ---------------- if auth_mode == "basic": header = request.META.get('HTTP_AUTHORIZATION') try: auth = header.split() except Exception: auth = "" uname = "" if len(auth) == 2: if auth[0].lower() == "basic": myenc = auth[1] myenc = bytes(myenc, encoding='utf-8') uname,pwd = str(base64.b64decode(myenc)).split(':') uname=uname[2:] # OIDC (OKTA , ...) elif auth_mode == "oauth2": login_field = get_configuration(appname="user", keyname="AUTH_FEDERATED_LOGIN_FIELD") uname = request.headers.get(login_field) email_field = get_configuration(appname="user", keyname="AUTH_FEDERATED_EMAIL_FIELD") email = request.headers.get(email_field) if debug_aaa == "yes": print("DEBUG_AAA - oauth2 login_field / uname ", login_field, uname) print("DEBUG_AAA - oauth2 email_field / email ", email_field, email) elif auth_mode == "local": if request.user.is_authenticated: uname = request.user.username aaa["is_anonymous"] = False if debug_aaa == "yes": print("DEBUG_AAA - auth_mode local - anonymous=False, request.user.username", uname) else: if debug_aaa == "yes": print("DEBUG_AAA - auth_mode local - no request.user.is_authenticated , no username") elif auth_mode == "unittest": uname = "unittest" aaa["is_anonymous"] = False elif auth_mode == "forced": # Force User from env/settings force_user = get_configuration(appname="env", keyname="CAVALIBA_FORCE_LOGIN") if debug_aaa == "yes": print("DEBUG_AAA - force_user ", force_user) if len(force_user) > 0: uname = force_user aaa["is_anonymous"] = False if debug_aaa == "yes": print("DEBUG_AAA - force_user ; is_anonymous=False") else: aaa["is_anonymous"] = True if debug_aaa == "yes": print("DEBUG_AAA - force_user ; is_anonymous=True. DONE.") return aaa elif auth_mode == "impersonate": uname = impersonate aaa["impersonate_by"] = impersonate_by if debug_aaa == "yes": print("DEBUG_AAA - uname=", uname) # ---------------- # anonymous # ---------------- # no uname : anonymous ! if not uname: aaa["is_anonymous"] = True if debug_aaa == "yes": print("DEBUG_AAA - no uname ; is_anonymous=True. DONE.") return aaa if len(uname) == 0: aaa["is_anonymous"] = True if debug_aaa == "yes": print("DEBUG_AAA - len(uname)=0 ; is_anonymous=True. DONE.") return aaa aaa["is_anonymous"] = False if debug_aaa == "yes": print("DEBUG_AAA - is_anonymous=False, uname=", uname) # ---------------- # username # ---------------- # remove domain from uname ? truncate_login = get_configuration(appname="user", keyname="AUTH_LOGIN_REMOVE_DOMAIN") if truncate_login == "yes": if '@' in uname: uname = re.sub("@(.*)$", '', uname) if debug_aaa == "yes": print("DEBUG_AAA - truncated domain name in login ; uname=", uname) # # ---------------- # # impersonate ? # # ---------------- # if uname == "admin": # impersonate = get_configuration(appname="user", keyname="SYSADMIN_IMPERSONATE") # if impersonate: # if debug_aaa == "yes": # print("DEBUG_AAA - impersonate: ", impersonate) # if len(impersonate) > 0: # uname = impersonate # aaa["impersonate"] = True # if debug_aaa == "yes": # print("DEBUG_AAA - impersonated=True, uname=", uname) # authenticated / visitor : register name aaa['username'] = uname # if debug_aaa == "yes": # print("DEBUG_AAA - uname post impersonate: ", uname) # ------------------------------- # check in DB > is_authenticated # ------------------------------- # default safe values aaa["is_visitor"] = True aaa["is_authenticated"] = False user = user_get_by_login(uname) if debug_aaa == "yes": print(f"DEBUG_AAA - user_get_by_login(): {user}") # JIT Just-in-Time provisioning if allowed and user not found jit = get_configuration(appname="user", keyname="AUTH_PROVISIONING") if debug_aaa == "yes": print("DEBUG_AAA - jit config: ", jit) # user in DB if user: aaa["is_visitor"] = False aaa['is_authenticated'] = True if debug_aaa == "yes": print("DEBUG_AAA - user found in DB ; visitor=False, authenticated=True") if not user.is_enabled: aaa['is_authenticated'] = False if debug_aaa == "yes": print("DEBUG_AAA - user disabled ; authenticated=False. DONE.") return aaa # user not in DB else: if debug_aaa == "yes": print("DEBUG_AAA - user NOT in DB , assuming visitor for now") # if uname == "admin": # aaa["is_visitor"] = False # aaa['is_authenticated'] = True # user = user_create({'login':uname}) # if debug_aaa == "yes": # print("DEBUG_AAA - uname is admin") if jit == "visitor": aaa["is_visitor"] = True aaa['is_authenticated'] = False # stay visitor, not authenticated (not in DB) if debug_aaa == "yes": print(f"DEBUG_AAA - jit ({jit}), visitor=True, authenticated=False") elif jit == "create" or jit == "sync": if not email: user = user_create({'login':uname}) else: user = user_create({'login':uname, 'email':email}) if user: log(INFO, aaa=aaa, data=f"JIT - ({jit}) get_aaa JIT user created in DB: {uname}") # not a visitor anymore, real user in DB aaa["is_visitor"] = False aaa['is_authenticated'] = True if debug_aaa == "yes": print(f"DEBUG_AAA - jit ({jit}) user created. visitor=False. authenticated=True") else: # failed to create in DB log(ERROR, aaa=aaa, data=f"JIT ({jit}) failed to create user in DB: {uname}") if debug_aaa == "yes": print(f"DEBUG_AAA - jit ({jit}) user not created. visitor=True. authenticated=False") aaa["is_visitor"] = True aaa['is_authenticated'] = True return aaa else: # unknown user, unittest, no JIT / no auto-create aaa["is_visitor"] = False aaa['is_authenticated'] = False if debug_aaa == "yes": print(f"DEBUG_AAA - not in DB, no jit or jit other: ({jit}). DONE") return aaa # --------------------- # user_id / user dict # --------------------- if user: aaa['userid'] = user.id #aaa['user'] = user # !!!! => Serialization KO when session/tasks / need conversion to dict dict_attributs = [ "login", "firstname", "lastname", "displayname", "email","mobile", "external_id", "is_enabled", "description", "want_notifications", "want_24", "want_email", "want_sms", "secondary_email", "secondary_mobile" ] # standard attibuts aaa['user'] = model_to_dict(user, fields=dict_attributs) if debug_aaa == "yes": print("DEBUG_AAA - aaa[user] attributes populated with DB content.") # ---------------- # Authorizations # ---------------- aaa["groups_direct"], aaa["groups_computed"], aaa["groups_indirect"] = get_groups_for_user(user, mode="name") aaa["groups"] = list(set(aaa["groups_direct"] + aaa["groups_computed"] + aaa["groups_indirect"])) # add perms & roles to aaa aaa["perms"] = role_get_permissions(aaa["groups"]) if debug_aaa == "yes": print("DEBUG_AAA - groups direct/computed/indirect and perms computed: ", len(aaa["groups"])) # ---------------- # admin # ---------------- # built-in user admin ? if aaa["username"] == "admin": aaa["is_admin"] = True if debug_aaa == "yes": print("DEBUG_AAA - username is admin => is_admin=True") # built-in role_admin ? #gobj_admin = role_get_by_name("role_admin", enabled_only=True) if 'role_admin' in aaa["groups"]: aaa["is_admin"] = True if debug_aaa == "yes": print("DEBUG_AAA - role_admin => is_admin=True") if aaa["is_admin"]: # give all groups, all perms aaa["perms"] = permission_all_keynames() aaa["groups"] = get_all_groups() if debug_aaa == "yes": print("DEBUG_AAA - all groups, perms for admin") if debug_aaa == "yes": print("DEBUG_AAA - get_aaa() end, DONE.") return aaa # ----------------------------------------------------------------------------- # helper functions # ----------------------------------------------------------------------------- def get_current_user(aaa): try: return user_get_by_login(aaa["username"]) except Exception: return # ------------------------------------ # get all groups # ------------------------------------ def get_all_groups(): allgroups = [] dbgroups = SireneGroup.objects.all() for g in dbgroups: allgroups.append(g.keyname) return allgroups # ------------------------------------------------- # get groups or groupnames and roles for aaa 'user' # ------------------------------------------------- def get_groups_for_user(user, mode="name"): if not isinstance(user, SireneUser): return [],[] direct = [] indirect = [] alldbgroups = SireneGroup.objects.filter(is_enabled = True)\ .prefetch_related("subgroups")\ .prefetch_related("users")\ .prefetch_related("permissions")\ .distinct() # 1. direct direct = list(SireneGroup.objects.filter(is_enabled = True, users__in=[user]).distinct()) # built-in: append "role_default" gobj = SireneGroup.objects.filter(is_enabled = True, is_role=True, keyname="role_default").first() if gobj: direct.append(gobj) # 2. computed groups from autogroup # TODO # direct.append() ... computed = list(SireneGroup.objects.filter(is_enabled = True, autogroup_users__in=[user]).distinct()) #direct = list(set(direct)) # 3. parent groups redo = 0 # get first level above direct groups for g0 in direct: for g1 in alldbgroups: if g0 in g1.subgroups.all(): if g1 not in direct: indirect.append(g1) redo = 1 # loop till no new indirect while redo > 0: redo = 0 for g1 in indirect: for g2 in alldbgroups: if g1 in g2.subgroups.all(): if g2 not in indirect: indirect.append(g2) redo = 1 indirect = list(set(indirect)) # convert to name only (e.g. JSON serialize to celery) if mode == "name": computed_name = [i.keyname for i in computed] indirect_name = [i.keyname for i in indirect] direct_name = [i.keyname for i in direct] return direct_name, computed_name, indirect_name # gobj else: return direct, computed, indirect