<s> '''<STR_LIT>''' <EOL> import io <EOL> import os <EOL> import sys <EOL> import subprocess <EOL> from . import subprocessio <EOL> import tempfile <EOL> from wsgiref . headers import Headers <EOL> import re <EOL> import urlparse <EOL> from collections import defaultdict <EOL> import time <EOL> import email . utils <EOL> import mimetypes <EOL> mimetypes . add_type ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mimetypes . add_type ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> from cStringIO import StringIO <EOL> import gzip <EOL> __version__ = ( <NUM_LIT:1> , <NUM_LIT:7> , <NUM_LIT:0> , <NUM_LIT:4> ) <EOL> class BaseWSGIClass ( object ) : <EOL> bufsize = <NUM_LIT> <EOL> gzip_response = False <EOL> canned_collection = { <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>' : '<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>' : "<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> } <EOL> def canned_handlers ( self , environ , start_response , code = '<STR_LIT>' , headers = [ ] ) : <EOL> '''<STR_LIT>''' <EOL> headerbase = [ ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) ] <EOL> if headers : <EOL> hObj = Headers ( headerbase ) <EOL> for header in headers : <EOL> hObj [ header [ <NUM_LIT:0> ] ] = '<STR_LIT>' . join ( header [ <NUM_LIT:1> : ] ) <EOL> start_response ( self . canned_collection [ code ] , headerbase ) <EOL> return [ '<STR_LIT>' ] <EOL> def package_response ( self , outIO , environ , start_response , headers = [ ] ) : <EOL> newheaders = headers <EOL> headers = [ ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> headersIface = Headers ( headers ) <EOL> for header in newheaders : <EOL> headersIface [ header [ <NUM_LIT:0> ] ] = '<STR_LIT>' . join ( header [ <NUM_LIT:1> : ] ) <EOL> if hasattr ( outIO , '<STR_LIT>' ) and '<STR_LIT>' in environ : <EOL> outIO . seek ( <NUM_LIT:0> ) <EOL> retobj = environ [ '<STR_LIT>' ] ( outIO , self . bufsize ) <EOL> elif hasattr ( outIO , '<STR_LIT>' ) : <EOL> outIO . seek ( <NUM_LIT:0> ) <EOL> retobj = iter ( lambda : outIO . read ( self . bufsize ) , '<STR_LIT>' ) <EOL> else : <EOL> retobj = outIO <EOL> start_response ( "<STR_LIT>" , headers ) <EOL> return retobj <EOL> class WSGIHandlerSelector ( BaseWSGIClass ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , WSGI_env_key = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> self . mappings = [ ] <EOL> self . WSGI_env_key = WSGI_env_key <EOL> def add ( self , path , default_handler = None , ** http_methods ) : <EOL> """<STR_LIT>""" <EOL> if default_handler : <EOL> methods = defaultdict ( lambda : default_handler , http_methods . copy ( ) ) <EOL> else : <EOL> methods = http_methods . copy ( ) <EOL> self . mappings . append ( ( re . compile ( path . decode ( '<STR_LIT:utf8>' ) ) , methods , ( path . find ( r'<STR_LIT>' ) > - <NUM_LIT:1> ) ) ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> path = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . decode ( '<STR_LIT:utf8>' ) <EOL> matches = None <EOL> handler = None <EOL> alternate_HTTP_verbs = set ( ) <EOL> query_string = ( environ . get ( '<STR_LIT>' ) or '<STR_LIT>' ) <EOL> path = urlparse . urljoin ( u'<STR_LIT:/>' , re . sub ( '<STR_LIT>' , '<STR_LIT:/>' , path . strip ( '<STR_LIT:/>' ) ) ) <EOL> if not path . startswith ( '<STR_LIT>' ) : <EOL> for _regex , _registered_methods , _use_query_string in self . mappings : <EOL> if _use_query_string : <EOL> matches = _regex . search ( path + '<STR_LIT:?>' + query_string ) <EOL> else : <EOL> matches = _regex . search ( path ) <EOL> if matches : <EOL> handler = _registered_methods [ environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ] <EOL> if handler : <EOL> break <EOL> else : <EOL> alternate_HTTP_verbs . update ( _registered_methods . keys ( ) ) <EOL> if handler : <EOL> environ [ '<STR_LIT>' ] = path . encode ( '<STR_LIT:utf8>' ) <EOL> mg = list ( environ . get ( '<STR_LIT>' ) or ( [ ] , { } ) ) <EOL> mg [ <NUM_LIT:0> ] = list ( mg [ <NUM_LIT:0> ] ) . append ( matches . groups ( ) ) , <EOL> mg [ <NUM_LIT:1> ] . update ( matches . groupdict ( ) ) <EOL> environ [ '<STR_LIT>' ] = tuple ( mg ) <EOL> return handler ( environ , start_response ) <EOL> elif alternate_HTTP_verbs : <EOL> return self . canned_handlers ( <EOL> environ , <EOL> start_response , <EOL> '<STR_LIT>' , <EOL> headers = [ ( '<STR_LIT>' , '<STR_LIT:U+002CU+0020>' . join ( alternate_HTTP_verbs ) ) ] <EOL> ) <EOL> else : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> class StaticWSGIServer ( BaseWSGIClass ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ** kw ) : <EOL> '''<STR_LIT>''' <EOL> self . __dict__ . update ( kw ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> selector_matches = ( environ . get ( '<STR_LIT>' ) or ( [ ] , { } ) ) [ <NUM_LIT:1> ] <EOL> if '<STR_LIT>' in selector_matches : <EOL> path_info = selector_matches [ '<STR_LIT>' ] . decode ( '<STR_LIT:utf8>' ) <EOL> else : <EOL> path_info = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) . decode ( '<STR_LIT:utf8>' ) <EOL> full_path = os . path . abspath ( os . path . join ( self . content_path , path_info . strip ( '<STR_LIT:/>' ) ) ) <EOL> _pp = os . path . abspath ( self . content_path ) <EOL> if not full_path . startswith ( _pp ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> if not os . path . isfile ( full_path ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> mtime = os . stat ( full_path ) . st_mtime <EOL> etag , last_modified = str ( mtime ) , email . utils . formatdate ( mtime ) <EOL> headers = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , email . utils . formatdate ( time . time ( ) ) ) , <EOL> ( '<STR_LIT>' , last_modified ) , <EOL> ( '<STR_LIT>' , etag ) <EOL> ] <EOL> headersIface = Headers ( headers ) <EOL> headersIface [ '<STR_LIT:Content-Type>' ] = mimetypes . guess_type ( full_path ) [ <NUM_LIT:0> ] or '<STR_LIT>' <EOL> if_modified = environ . get ( '<STR_LIT>' ) <EOL> if if_modified and ( email . utils . parsedate ( if_modified ) >= email . utils . parsedate ( last_modified ) ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' , headers ) <EOL> if_none = environ . get ( '<STR_LIT>' ) <EOL> if if_none and ( if_none == '<STR_LIT:*>' or etag in if_none ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' , headers ) <EOL> file_like = open ( full_path , '<STR_LIT:rb>' ) <EOL> return self . package_response ( file_like , environ , start_response , headers ) <EOL> class GitHTTPBackendBase ( BaseWSGIClass ) : <EOL> git_folder_signature = set ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:info>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> repo_auto_create = True <EOL> def has_access ( self , ** kw ) : <EOL> '''<STR_LIT>''' <EOL> return True <EOL> def basic_checks ( self , dataObj , environ , start_response ) : <EOL> '''<STR_LIT>''' <EOL> selector_matches = ( environ . get ( '<STR_LIT>' ) or ( [ ] , { } ) ) [ <NUM_LIT:1> ] <EOL> git_command = selector_matches . get ( '<STR_LIT>' ) or '<STR_LIT>' <EOL> if git_command not in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> repo_path = os . path . abspath ( <EOL> os . path . join ( <EOL> self . content_path , <EOL> ( selector_matches . get ( '<STR_LIT>' ) or '<STR_LIT>' ) . decode ( '<STR_LIT:utf8>' ) . strip ( '<STR_LIT:/>' ) . strip ( '<STR_LIT:\\>' ) <EOL> ) <EOL> ) <EOL> _pp = os . path . abspath ( self . content_path ) <EOL> if not repo_path . startswith ( _pp ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> if not self . has_access ( <EOL> environ = environ , <EOL> repo_path = repo_path , <EOL> git_command = git_command <EOL> ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> try : <EOL> files = os . listdir ( repo_path ) <EOL> except : <EOL> files = [ ] <EOL> if not self . git_folder_signature . issubset ( [ i . lower ( ) for i in files ] ) : <EOL> if not ( self . repo_auto_create and git_command == '<STR_LIT>' ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> else : <EOL> _pf = _pp <EOL> _dirs = repo_path [ len ( _pp ) : ] . strip ( os . sep ) . split ( os . sep ) or [ '<STR_LIT>' ] <EOL> for _dir in _dirs : <EOL> _pf = os . path . join ( _pf , _dir ) <EOL> if not os . path . exists ( _pf ) : <EOL> try : <EOL> os . makedirs ( repo_path ) <EOL> except : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> break <EOL> elif not os . path . isdir ( _pf ) or self . git_folder_signature . issubset ( [ i . lower ( ) for i in os . listdir ( _pf ) ] ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> if subprocess . call ( '<STR_LIT>' % repo_path , shell = True ) : <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> dataObj [ '<STR_LIT>' ] = git_command <EOL> dataObj [ '<STR_LIT>' ] = repo_path <EOL> return None <EOL> class GitHTTPBackendInfoRefs ( GitHTTPBackendBase ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , ** kw ) : <EOL> '''<STR_LIT>''' <EOL> self . __dict__ . update ( kw ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> dataObj = { } <EOL> answer = self . basic_checks ( dataObj , environ , start_response ) <EOL> if answer : <EOL> return answer <EOL> git_command = dataObj [ '<STR_LIT>' ] <EOL> repo_path = dataObj [ '<STR_LIT>' ] <EOL> smart_server_advert = '<STR_LIT>' % git_command <EOL> try : <EOL> out = subprocessio . SubprocessIOChunker ( <EOL> r'<STR_LIT>' % ( git_command [ <NUM_LIT:4> : ] , repo_path ) , <EOL> starting_values = [ str ( hex ( len ( smart_server_advert ) + <NUM_LIT:4> ) [ <NUM_LIT:2> : ] . rjust ( <NUM_LIT:4> , '<STR_LIT:0>' ) + smart_server_advert + '<STR_LIT>' ) ] , <EOL> env = dict ( CODE_REMOTE_USER = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ) <EOL> except ( EnvironmentError ) as e : <EOL> environ [ '<STR_LIT>' ] . write ( str ( e ) ) <EOL> return self . canned_handlers ( environ , start_response , '<STR_LIT>' ) <EOL> headers = [ ( '<STR_LIT>' , '<STR_LIT>' % str ( git_command ) ) ] <EOL> return self . package_response ( <EOL> out , <EOL> environ , <EOL> start_response , <EOL> headers ) <EOL> class GitHTTPBackendSmartHTTP ( GitHTTPBackendBase ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , ** kw ) : <EOL> '''<STR_LIT>''' <EOL> self . __dict__ . update ( kw ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> dataObj = { } <EOL> answer = self . basic_checks ( dataObj , environ , start_response ) <EOL> if answer : <EOL> return answer <EOL> git_command = dataObj [ '<STR_LIT>' ] <EOL> repo_path = dataObj [ '<STR_LIT>' ] <EOL> try : <EOL> _l = int ( environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> except : <EOL> _l = None <EOL> wsgi_version = environ . get ( '<STR_LIT>' , ( <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> if wsgi_version [ <NUM_LIT:0> ] >= <NUM_LIT:1> and wsgi_version [ <NUM_LIT:1> ] >= <NUM_LIT:1> : <EOL> wsgi_input_has_EOF = True <EOL> else : <EOL> wsgi_input_has_EOF = False <EOL> _i = environ . get ( '<STR_LIT>' ) <EOL> if wsgi_input_has_EOF : <EOL> stdin = _i <EOL> else : <EOL> if _l > self . bufsize : <EOL> bs = self . bufsize <EOL> btr = _l <EOL> stdin = tempfile . TemporaryFile ( ) <EOL> while btr >= bs : <EOL> stdin . write ( _i . read ( bs ) ) <EOL> btr -= bs <EOL> stdin . write ( _i . read ( btr ) ) <EOL> stdin . flush ( ) <EOL> stdin . seek ( <NUM_LIT:0> ) <EOL> else : <EOL> stdin = _i . read ( _l ) <EOL> cmd = r'<STR_LIT>' % ( git_command [ <NUM_LIT:4> : ] , repo_path ) <EOL> _p = subprocess . Popen ( cmd , <EOL> bufsize = - <NUM_LIT:1> , <EOL> shell = True , <EOL> stdin = subprocess . PIPE , <EOL> stdout = subprocess . PIPE , <EOL> env = dict ( CODE_REMOTE_USER = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ) <EOL> if isinstance ( stdin , file ) : <EOL> out , _ = _p . communicate ( stdin . read ( ) ) <EOL> else : <EOL> try : <EOL> fp = gzip . GzipFile ( fileobj = StringIO ( stdin ) ) <EOL> out , _ = _p . communicate ( fp . read ( ) ) <EOL> except IOError : <EOL> out , _ = _p . communicate ( stdin ) <EOL> out = StringIO ( out ) <EOL> if git_command == u'<STR_LIT>' : <EOL> subprocess . call ( u'<STR_LIT>' % repo_path , shell = True ) <EOL> headers = [ ( '<STR_LIT>' , '<STR_LIT>' % git_command . encode ( '<STR_LIT:utf8>' ) ) ] <EOL> return self . package_response ( <EOL> out , <EOL> environ , <EOL> start_response , <EOL> headers ) <EOL> def assemble_WSGI_git_app ( * args , ** kw ) : <EOL> '''<STR_LIT>''' <EOL> default_options = [ <EOL> [ '<STR_LIT>' , '<STR_LIT:.>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ] <EOL> args = list ( args ) <EOL> options = dict ( default_options ) <EOL> options . update ( kw ) <EOL> while default_options and args : <EOL> _d = default_options . pop ( <NUM_LIT:0> ) <EOL> _a = args . pop ( <NUM_LIT:0> ) <EOL> options [ _d [ <NUM_LIT:0> ] ] = _a <EOL> options [ '<STR_LIT>' ] = os . path . abspath ( options [ '<STR_LIT>' ] . decode ( '<STR_LIT:utf8>' ) ) <EOL> options [ '<STR_LIT>' ] = options [ '<STR_LIT>' ] . decode ( '<STR_LIT:utf8>' ) <EOL> selector = WSGIHandlerSelector ( ) <EOL> generic_handler = StaticWSGIServer ( ** options ) <EOL> git_inforefs_handler = GitHTTPBackendInfoRefs ( ** options ) <EOL> git_rpc_handler = GitHTTPBackendSmartHTTP ( ** options ) <EOL> if options [ '<STR_LIT>' ] : <EOL> marker_regex = r'<STR_LIT>' + options [ '<STR_LIT>' ] + '<STR_LIT:)>' <EOL> else : <EOL> marker_regex = '<STR_LIT>' <EOL> selector . add ( <EOL> marker_regex + r'<STR_LIT>' , <EOL> GET = git_inforefs_handler , <EOL> HEAD = git_inforefs_handler <EOL> ) <EOL> selector . add ( <EOL> marker_regex + r'<STR_LIT>' , <EOL> POST = git_rpc_handler <EOL> ) <EOL> selector . add ( <EOL> marker_regex + r'<STR_LIT>' , <EOL> GET = generic_handler , <EOL> HEAD = generic_handler ) <EOL> return selector <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> _help = r'''<STR_LIT>''' <EOL> import sys <EOL> command_options = { <EOL> '<STR_LIT>' : '<STR_LIT:.>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' <EOL> } <EOL> lastKey = None <EOL> for item in sys . argv : <EOL> if item . startswith ( '<STR_LIT>' ) : <EOL> command_options [ item [ <NUM_LIT:2> : ] ] = True <EOL> lastKey = item [ <NUM_LIT:2> : ] <EOL> elif lastKey : <EOL> command_options [ lastKey ] = item . strip ( '<STR_LIT:">' ) . strip ( "<STR_LIT:'>" ) <EOL> lastKey = None <EOL> content_path = os . path . abspath ( command_options [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in command_options : <EOL> print _help <EOL> else : <EOL> app = assemble_WSGI_git_app ( <EOL> content_path = content_path , <EOL> uri_marker = command_options [ '<STR_LIT>' ] , <EOL> performance_settings = { <EOL> '<STR_LIT>' : True <EOL> } <EOL> ) <EOL> from cherrypy import wsgiserver <EOL> httpd = wsgiserver . CherryPyWSGIServer ( ( '<STR_LIT>' , int ( command_options [ '<STR_LIT:port>' ] ) ) , app ) <EOL> if command_options [ '<STR_LIT>' ] : <EOL> _s = '<STR_LIT>' % command_options [ '<STR_LIT>' ] <EOL> example_URI = '''<STR_LIT>''' % ( <EOL> command_options [ '<STR_LIT:port>' ] , <EOL> command_options [ '<STR_LIT>' ] , <EOL> command_options [ '<STR_LIT>' ] ) <EOL> else : <EOL> _s = '<STR_LIT>' <EOL> example_URI = '<STR_LIT>' % ( command_options [ '<STR_LIT:port>' ] ) <EOL> print '''<STR_LIT>''' % ( command_options [ '<STR_LIT:port>' ] , content_path , _s , example_URI ) <EOL> try : <EOL> httpd . start ( ) <EOL> except KeyboardInterrupt : <EOL> pass <EOL> finally : <EOL> httpd . stop ( ) </s>
