<s> """<STR_LIT>""" <EOL> import sys , os , socket , time , subprocess , logging <EOL> _log = logging . getLogger ( "<STR_LIT>" ) <EOL> def quote ( s ) : <EOL> '''<STR_LIT>''' <EOL> return '<STR_LIT:">' + s . replace ( '<STR_LIT:\\>' , '<STR_LIT>' ) . replace ( '<STR_LIT:">' , '<STR_LIT>' ) + '<STR_LIT:">' <EOL> def mkdirs ( path ) : <EOL> """<STR_LIT>""" <EOL> if not os . access ( os . path . join ( path , os . path . curdir ) , os . F_OK ) : <EOL> _log . debug ( "<STR_LIT>" , path ) <EOL> os . mkdir ( path ) <EOL> return path <EOL> def delete_directory_content ( path ) : <EOL> for dirpath , dirnames , filenames in os . walk ( path , topdown = False ) : <EOL> for n in filenames : <EOL> _log . info ( "<STR_LIT>" , os . path . join ( dirpath , n ) ) <EOL> os . remove ( os . path . join ( dirpath , n ) ) <EOL> for n in dirnames : <EOL> _log . info ( "<STR_LIT>" , os . path . join ( dirpath , n ) ) <EOL> os . rmdir ( os . path . join ( dirpath , n ) ) <EOL> LOCALHOST = '<STR_LIT:127.0.0.1>' <EOL> def find_available_tcp_port ( host = LOCALHOST ) : <EOL> s = socket . socket ( ) <EOL> s . bind ( ( host , <NUM_LIT:0> ) ) <EOL> port = s . getsockname ( ) [ <NUM_LIT:1> ] <EOL> s . close ( ) <EOL> _log . info ( "<STR_LIT>" , port ) <EOL> return port <EOL> class Slapd : <EOL> """<STR_LIT>""" <EOL> _log = logging . getLogger ( "<STR_LIT>" ) <EOL> PATH_TMPDIR = "<STR_LIT>" <EOL> PATH_SBINDIR = "<STR_LIT>" <EOL> PATH_BINDIR = "<STR_LIT>" <EOL> PATH_SCHEMA_CORE = "<STR_LIT>" <EOL> PATH_LDAPADD = os . path . join ( PATH_BINDIR , "<STR_LIT>" ) <EOL> PATH_LDAPSEARCH = os . path . join ( PATH_BINDIR , "<STR_LIT>" ) <EOL> PATH_SLAPD = os . path . join ( PATH_SBINDIR , "<STR_LIT>" ) <EOL> PATH_SLAPTEST = os . path . join ( PATH_SBINDIR , "<STR_LIT>" ) <EOL> def check_paths ( cls ) : <EOL> """<STR_LIT>""" <EOL> for name , path in ( <EOL> ( "<STR_LIT>" , cls . PATH_SLAPD ) , <EOL> ( "<STR_LIT>" , cls . PATH_LDAPADD ) , <EOL> ( "<STR_LIT>" , cls . PATH_LDAPSEARCH ) , <EOL> ) : <EOL> cls . _log . debug ( "<STR_LIT>" , name , path ) <EOL> if not os . access ( path , os . X_OK ) : <EOL> cls . _log . warn ( "<STR_LIT>" , name , path ) <EOL> check_paths = classmethod ( check_paths ) <EOL> def __init__ ( self ) : <EOL> self . _config = [ ] <EOL> self . _proc = None <EOL> self . _port = <NUM_LIT:0> <EOL> self . _tmpdir = self . PATH_TMPDIR <EOL> self . _dn_suffix = "<STR_LIT>" <EOL> self . _root_cn = "<STR_LIT>" <EOL> self . _root_password = "<STR_LIT:password>" <EOL> self . _slapd_debug_level = <NUM_LIT:0> <EOL> def set_port ( self , port ) : <EOL> self . _port = port <EOL> def set_dn_suffix ( self , dn ) : <EOL> self . _dn_suffix = dn <EOL> def set_root_cn ( self , cn ) : <EOL> self . _root_cn = cn <EOL> def set_root_password ( self , pw ) : <EOL> self . _root_password = pw <EOL> def set_tmpdir ( self , path ) : <EOL> self . _tmpdir = path <EOL> def set_slapd_debug_level ( self , level ) : <EOL> self . _slapd_debug_level = level <EOL> def set_debug ( self ) : <EOL> self . _log . setLevel ( logging . DEBUG ) <EOL> self . set_slapd_debug_level ( '<STR_LIT>' ) <EOL> def get_url ( self ) : <EOL> return "<STR_LIT>" % self . get_address ( ) <EOL> def get_address ( self ) : <EOL> if self . _port == <NUM_LIT:0> : <EOL> self . _port = find_available_tcp_port ( LOCALHOST ) <EOL> return ( LOCALHOST , self . _port ) <EOL> def get_dn_suffix ( self ) : <EOL> return self . _dn_suffix <EOL> def get_root_dn ( self ) : <EOL> return "<STR_LIT>" + self . _root_cn + "<STR_LIT:U+002C>" + self . get_dn_suffix ( ) <EOL> def get_root_password ( self ) : <EOL> return self . _root_password <EOL> def get_tmpdir ( self ) : <EOL> return self . _tmpdir <EOL> def __del__ ( self ) : <EOL> self . stop ( ) <EOL> def configure ( self , cfg ) : <EOL> """<STR_LIT>""" <EOL> cfg . append ( "<STR_LIT>" + quote ( self . PATH_SCHEMA_CORE ) ) <EOL> cfg . append ( "<STR_LIT>" ) <EOL> ldif_dir = mkdirs ( os . path . join ( self . get_tmpdir ( ) , "<STR_LIT>" ) ) <EOL> delete_directory_content ( ldif_dir ) <EOL> cfg . append ( "<STR_LIT>" ) <EOL> cfg . append ( "<STR_LIT>" + quote ( ldif_dir ) ) <EOL> cfg . append ( "<STR_LIT>" + quote ( self . get_dn_suffix ( ) ) ) <EOL> cfg . append ( "<STR_LIT>" + quote ( self . get_root_dn ( ) ) ) <EOL> cfg . append ( "<STR_LIT>" + quote ( self . get_root_password ( ) ) ) <EOL> def _write_config ( self ) : <EOL> """<STR_LIT>""" <EOL> path = os . path . join ( self . _tmpdir , "<STR_LIT>" ) <EOL> ldif_dir = mkdirs ( self . _tmpdir ) <EOL> if os . access ( path , os . F_OK ) : <EOL> self . _log . debug ( "<STR_LIT>" , path ) <EOL> os . remove ( path ) <EOL> self . _log . debug ( "<STR_LIT>" , path ) <EOL> file ( path , "<STR_LIT:w>" ) . writelines ( [ line + "<STR_LIT:\n>" for line in self . _config ] ) <EOL> return path <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _proc is None : <EOL> ok = False <EOL> config_path = None <EOL> try : <EOL> self . configure ( self . _config ) <EOL> self . _test_configuration ( ) <EOL> self . _start_slapd ( ) <EOL> self . _wait_for_slapd ( ) <EOL> ok = True <EOL> self . _log . debug ( "<STR_LIT>" , self . get_url ( ) ) <EOL> self . started ( ) <EOL> finally : <EOL> if not ok : <EOL> if config_path : <EOL> try : os . remove ( config_path ) <EOL> except os . error : pass <EOL> if self . _proc : <EOL> self . stop ( ) <EOL> def _start_slapd ( self ) : <EOL> config_path = self . _write_config ( ) <EOL> self . _log . info ( "<STR_LIT>" ) <EOL> self . _proc = subprocess . Popen ( [ self . PATH_SLAPD , <EOL> "<STR_LIT>" , config_path , <EOL> "<STR_LIT>" , self . get_url ( ) , <EOL> "<STR_LIT>" , str ( self . _slapd_debug_level ) , <EOL> ] ) <EOL> self . _proc_config = config_path <EOL> def _wait_for_slapd ( self ) : <EOL> s = socket . socket ( ) <EOL> while <NUM_LIT:1> : <EOL> if self . _proc . poll ( ) is not None : <EOL> self . _stopped ( ) <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> try : <EOL> self . _log . debug ( "<STR_LIT>" , repr ( self . get_address ( ) ) ) <EOL> s . connect ( self . get_address ( ) ) <EOL> s . close ( ) <EOL> return <EOL> except socket . error : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _proc is not None : <EOL> self . _log . debug ( "<STR_LIT>" ) <EOL> if hasattr ( self . _proc , '<STR_LIT>' ) : <EOL> self . _proc . terminate ( ) <EOL> else : <EOL> import posix , signal <EOL> posix . kill ( self . _proc . pid , signal . SIGHUP ) <EOL> self . wait ( ) <EOL> def restart ( self ) : <EOL> """<STR_LIT>""" <EOL> self . stop ( ) <EOL> self . start ( ) <EOL> def wait ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _proc : <EOL> self . _proc . wait ( ) <EOL> self . _stopped ( ) <EOL> def _stopped ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _proc is not None : <EOL> self . _log . info ( "<STR_LIT>" ) <EOL> self . _proc = None <EOL> try : <EOL> os . remove ( self . _proc_config ) <EOL> except os . error : <EOL> self . _log . debug ( "<STR_LIT>" , self . _proc_config ) <EOL> def _test_configuration ( self ) : <EOL> config_path = self . _write_config ( ) <EOL> try : <EOL> self . _log . debug ( "<STR_LIT>" ) <EOL> verboseflag = "<STR_LIT>" <EOL> if self . _log . isEnabledFor ( logging . DEBUG ) : <EOL> verboseflag = "<STR_LIT>" <EOL> p = subprocess . Popen ( [ <EOL> self . PATH_SLAPTEST , <EOL> verboseflag , <EOL> "<STR_LIT>" , config_path <EOL> ] ) <EOL> if p . wait ( ) != <NUM_LIT:0> : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> self . _log . debug ( "<STR_LIT>" ) <EOL> finally : <EOL> os . remove ( config_path ) <EOL> def ldapadd ( self , ldif , extra_args = [ ] ) : <EOL> """<STR_LIT>""" <EOL> self . _log . debug ( "<STR_LIT>" , repr ( ldif ) ) <EOL> p = subprocess . Popen ( [ self . PATH_LDAPADD , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , self . get_root_dn ( ) , <EOL> "<STR_LIT>" , self . get_root_password ( ) , <EOL> "<STR_LIT>" , self . get_url ( ) ] + extra_args , <EOL> stdin = subprocess . PIPE , stdout = subprocess . PIPE ) <EOL> p . communicate ( ldif ) <EOL> if p . wait ( ) != <NUM_LIT:0> : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def ldapsearch ( self , base = None , filter = '<STR_LIT>' , attrs = [ ] , <EOL> scope = '<STR_LIT>' , extra_args = [ ] ) : <EOL> if base is None : base = self . get_dn_suffix ( ) <EOL> self . _log . debug ( "<STR_LIT>" , repr ( filter ) ) <EOL> p = subprocess . Popen ( [ self . PATH_LDAPSEARCH , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , self . get_root_dn ( ) , <EOL> "<STR_LIT>" , self . get_root_password ( ) , <EOL> "<STR_LIT>" , self . get_url ( ) , <EOL> "<STR_LIT>" , base , <EOL> "<STR_LIT>" , scope , <EOL> "<STR_LIT>" , <EOL> ] + extra_args + [ filter ] + attrs , <EOL> stdout = subprocess . PIPE ) <EOL> output = p . communicate ( ) [ <NUM_LIT:0> ] <EOL> if p . wait ( ) != <NUM_LIT:0> : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> lines = [ ] <EOL> for l in output . split ( '<STR_LIT:\n>' ) : <EOL> if l . startswith ( '<STR_LIT:U+0020>' ) : <EOL> lines [ - <NUM_LIT:1> ] = lines [ - <NUM_LIT:1> ] + l [ <NUM_LIT:1> : ] <EOL> elif l == '<STR_LIT>' and lines and lines [ - <NUM_LIT:1> ] == '<STR_LIT>' : <EOL> pass <EOL> else : <EOL> lines . append ( l ) <EOL> lines = [ l for l in lines if not l . startswith ( "<STR_LIT:#>" ) ] <EOL> if lines and lines [ <NUM_LIT:0> ] == '<STR_LIT>' : del lines [ <NUM_LIT:0> ] <EOL> if not lines or lines [ <NUM_LIT:0> ] != '<STR_LIT>' : <EOL> raise RuntimeError ( "<STR_LIT>" + repr ( lines [ : <NUM_LIT:1> ] ) ) <EOL> del lines [ <NUM_LIT:0> ] <EOL> if lines and lines [ <NUM_LIT:0> ] == '<STR_LIT>' : del lines [ <NUM_LIT:0> ] <EOL> if lines and lines [ - <NUM_LIT:1> ] != '<STR_LIT>' : lines . append ( '<STR_LIT>' ) <EOL> objects = [ ] <EOL> obj = [ ] <EOL> for line in lines : <EOL> if line == '<STR_LIT>' : <EOL> if obj [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] != '<STR_LIT>' : <EOL> raise RuntimeError ( "<STR_LIT>" , repr ( obj ) ) <EOL> objects . append ( ( obj [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , obj [ <NUM_LIT:1> : ] ) ) <EOL> obj = [ ] <EOL> else : <EOL> attr , value = line . split ( '<STR_LIT::>' , <NUM_LIT:2> ) <EOL> if value . startswith ( '<STR_LIT>' ) : <EOL> value = base64 . decodestring ( value [ <NUM_LIT:2> : ] ) <EOL> elif value . startswith ( '<STR_LIT:U+0020>' ) : <EOL> value = value [ <NUM_LIT:1> : ] <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" + repr ( line ) ) <EOL> obj . append ( ( attr , value ) ) <EOL> assert obj == [ ] <EOL> return objects <EOL> def started ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . get_dn_suffix ( ) . startswith ( "<STR_LIT>" ) <EOL> suffix_dc = self . get_dn_suffix ( ) . split ( '<STR_LIT:U+002C>' ) [ <NUM_LIT:0> ] [ <NUM_LIT:3> : ] <EOL> assert self . get_root_dn ( ) . startswith ( "<STR_LIT>" ) <EOL> assert self . get_root_dn ( ) . endswith ( "<STR_LIT:U+002C>" + self . get_dn_suffix ( ) ) <EOL> root_cn = self . get_root_dn ( ) . split ( '<STR_LIT:U+002C>' ) [ <NUM_LIT:0> ] [ <NUM_LIT:3> : ] <EOL> self . _log . debug ( "<STR_LIT>" , <EOL> self . get_dn_suffix ( ) , <EOL> self . get_root_dn ( ) ) <EOL> self . ldapadd ( "<STR_LIT:\n>" . join ( [ <EOL> '<STR_LIT>' + self . get_dn_suffix ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' + suffix_dc , <EOL> '<STR_LIT>' + suffix_dc , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' + self . get_root_dn ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' + root_cn , <EOL> '<STR_LIT>' <EOL> ] ) ) <EOL> Slapd . check_paths ( ) <EOL> if __name__ == '<STR_LIT:__main__>' and sys . argv == [ '<STR_LIT>' ] : <EOL> logging . basicConfig ( level = logging . DEBUG ) <EOL> slapd = Slapd ( ) <EOL> print ( "<STR_LIT>" ) <EOL> slapd . start ( ) <EOL> print ( "<STR_LIT>" ) <EOL> for dn , attrs in slapd . ldapsearch ( ) : <EOL> print ( "<STR_LIT>" + dn ) <EOL> for name , val in attrs : <EOL> print ( name + "<STR_LIT>" + val ) <EOL> print ( "<STR_LIT>" ) <EOL> print ( slapd . get_url ( ) ) <EOL> slapd . wait ( ) </s>
