<s> from six import viewitems <EOL> from . interface import BUIhandler , BUIuser , BUIloader <EOL> from ... _compat import ConfigParser <EOL> import ssl <EOL> try : <EOL> from ldap3 import Server , Connection , Tls , ALL , RESTARTABLE , AUTO_BIND_TLS_BEFORE_BIND , AUTO_BIND_NONE , SIMPLE <EOL> except ImportError : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> class LdapLoader ( BUIloader ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app = None , handler = None ) : <EOL> """<STR_LIT>""" <EOL> self . app = app <EOL> conf = self . app . config [ '<STR_LIT>' ] <EOL> defaults = { <EOL> '<STR_LIT:host>' : '<STR_LIT:localhost>' , <EOL> '<STR_LIT:port>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:none>' , <EOL> '<STR_LIT:version>' : None , <EOL> '<STR_LIT>' : None <EOL> } <EOL> mapping = { <EOL> '<STR_LIT:host>' : '<STR_LIT:host>' , <EOL> '<STR_LIT:port>' : '<STR_LIT:port>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:version>' : '<STR_LIT:version>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> c = ConfigParser . ConfigParser ( defaults ) <EOL> with open ( conf ) as fp : <EOL> c . readfp ( fp ) <EOL> try : <EOL> handler . priority = c . getint ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> except : <EOL> pass <EOL> for ( opt , key ) in viewitems ( mapping ) : <EOL> try : <EOL> setattr ( self , opt , c . get ( '<STR_LIT>' , key ) ) <EOL> except ConfigParser . NoOptionError as e : <EOL> self . builogger . info ( str ( e ) ) <EOL> except ConfigParser . NoSectionError as e : <EOL> self . builogger . error ( str ( e ) ) <EOL> if self . validate and self . validate . lower ( ) in [ '<STR_LIT:none>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . validate = getattr ( ssl , '<STR_LIT>' . format ( self . validate . upper ( ) ) ) <EOL> else : <EOL> self . validate = None <EOL> if self . version and self . version in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> self . version = getattr ( ssl , '<STR_LIT>' . format ( self . version ) ) <EOL> else : <EOL> self . version = None <EOL> self . tls = None <EOL> self . ssl = False <EOL> self . auto_bind = AUTO_BIND_NONE <EOL> if self . encryption == '<STR_LIT>' : <EOL> self . ssl = True <EOL> elif self . encryption == '<STR_LIT>' : <EOL> self . tls = Tls ( local_certificate_file = self . cafile , validate = self . validate , version = self . version ) <EOL> self . auto_bind = AUTO_BIND_TLS_BEFORE_BIND <EOL> if self . port : <EOL> try : <EOL> self . port = int ( self . port ) <EOL> except ValueError : <EOL> self . builogger . error ( '<STR_LIT>' ) <EOL> self . port = None <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . host ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . port ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . encryption ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . filt ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . base ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . attr ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . binddn ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( '<STR_LIT>' if self . bindpw else '<STR_LIT:None>' ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( self . tls ) ) <EOL> try : <EOL> self . server = Server ( host = self . host , port = self . port , use_ssl = self . ssl , get_info = ALL , tls = self . tls ) <EOL> self . builogger . debug ( '<STR_LIT>' . format ( str ( self . server ) ) ) <EOL> if self . binddn : <EOL> self . ldap = Connection ( self . server , user = self . binddn , password = self . bindpw , raise_exceptions = True , client_strategy = RESTARTABLE , auto_bind = self . auto_bind , authentication = SIMPLE ) <EOL> else : <EOL> self . ldap = Connection ( self . server , raise_exceptions = True , client_strategy = RESTARTABLE , auto_bind = self . auto_bind ) <EOL> with self . ldap : <EOL> self . builogger . debug ( '<STR_LIT>' . format ( str ( self . ldap ) ) ) <EOL> self . builogger . info ( '<STR_LIT>' ) <EOL> return <EOL> raise Exception ( '<STR_LIT>' ) <EOL> except Exception as e : <EOL> self . builogger . error ( '<STR_LIT>' . format ( str ( e ) ) ) <EOL> self . server = None <EOL> self . ldap = None <EOL> def __exit__ ( self , exc_type , exc_value , traceback ) : <EOL> """<STR_LIT>""" <EOL> if self . ldap and self . ldap . bound : <EOL> self . ldap . unbind ( ) <EOL> def fetch ( self , searchval = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> if self . filt : <EOL> query = self . filt . format ( self . attr , searchval ) <EOL> else : <EOL> query = '<STR_LIT>' . format ( self . attr , searchval ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( query , self . base ) ) <EOL> r = None <EOL> with self . ldap : <EOL> self . builogger . debug ( '<STR_LIT>' . format ( str ( self . ldap ) ) ) <EOL> self . ldap . search ( self . base , query , attributes = [ '<STR_LIT>' , self . attr ] ) <EOL> r = self . ldap . response <EOL> if not r : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> except Exception as e : <EOL> self . builogger . error ( '<STR_LIT>' . format ( str ( e ) ) ) <EOL> return None <EOL> for record in r : <EOL> attrs = record [ '<STR_LIT>' ] <EOL> if self . attr in attrs and searchval in attrs [ self . attr ] : <EOL> self . builogger . info ( '<STR_LIT>' . format ( record [ '<STR_LIT>' ] ) ) <EOL> return { '<STR_LIT>' : record [ '<STR_LIT>' ] , '<STR_LIT>' : attrs [ '<STR_LIT>' ] [ <NUM_LIT:0> ] } <EOL> def check ( self , dn = None , passwd = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> with Connection ( self . server , user = '<STR_LIT>' . format ( dn ) , password = passwd , raise_exceptions = True , auto_bind = self . auto_bind , authentication = SIMPLE ) as l : <EOL> self . builogger . debug ( '<STR_LIT>' . format ( str ( l ) ) ) <EOL> self . builogger . info ( '<STR_LIT>' . format ( dn ) ) <EOL> return l . bind ( ) <EOL> except Exception as e : <EOL> self . builogger . error ( '<STR_LIT>' . format ( dn , str ( e ) ) ) <EOL> self . builogger . error ( '<STR_LIT>' . format ( dn ) ) <EOL> return False <EOL> class UserHandler ( BUIhandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app = None , auth = None ) : <EOL> """<STR_LIT>""" <EOL> self . ldap = LdapLoader ( app , self ) <EOL> self . users = { } <EOL> def user ( self , name = None ) : <EOL> """<STR_LIT>""" <EOL> if name not in self . users : <EOL> self . users [ name ] = LdapUser ( self . ldap , name ) <EOL> return self . users [ name ] <EOL> class LdapUser ( BUIuser ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ldap = None , name = None ) : <EOL> """<STR_LIT>""" <EOL> self . active = False <EOL> self . authenticated = False <EOL> self . ldap = ldap <EOL> self . name = name <EOL> found = self . ldap . fetch ( name ) <EOL> if found : <EOL> self . id = found [ '<STR_LIT>' ] <EOL> self . active = True <EOL> def login ( self , name = None , passwd = None ) : <EOL> """<STR_LIT>""" <EOL> if self . ldap . fetch ( name ) : <EOL> self . authenticated = self . ldap . check ( self . id , passwd ) <EOL> return self . authenticated <EOL> else : <EOL> self . authenticated = False <EOL> self . active = False <EOL> return False <EOL> @ property <EOL> def is_active ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . active <EOL> @ property <EOL> def is_authenticated ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . authenticated <EOL> def get_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . name </s>
