<s> import re <EOL> import time <EOL> import os <EOL> import select <EOL> from os import access <EOL> from os . path import join , exists , getmtime , getsize <EOL> from urllib import unquote <EOL> from BaseHTTPServer import BaseHTTPRequestHandler as _ <EOL> from libs . git import Git <EOL> def format_date_time ( timestamp ) : <EOL> year , month , day , hh , mm , ss , wd , y , z = time . gmtime ( timestamp ) <EOL> return "<STR_LIT>" % ( <EOL> _ . weekdayname [ wd ] , day , _ . monthname [ month ] , year , hh , mm , ss <EOL> ) <EOL> def callback ( p ) : <EOL> ofd = p . stdout . fileno ( ) <EOL> efd = p . stderr . fileno ( ) <EOL> while True : <EOL> r_ready , w_ready , x_ready = select . select ( [ ofd , efd ] , [ ] , [ ] , <NUM_LIT:30> ) <EOL> if ofd in r_ready : <EOL> data = os . read ( ofd , <NUM_LIT> ) <EOL> if not data : <EOL> break <EOL> yield data <EOL> if efd in r_ready : <EOL> data = os . read ( efd , <NUM_LIT> ) <EOL> yield data <EOL> break <EOL> output , err = p . communicate ( ) <EOL> if output : <EOL> yield output <EOL> if err : <EOL> yield err <EOL> class GHTTPServer ( object ) : <EOL> VALID_SERVICE_TYPES = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> SERVICES = [ <EOL> [ "<STR_LIT:POST>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) , '<STR_LIT>' ] , <EOL> [ "<STR_LIT:POST>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) , '<STR_LIT>' ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> [ "<STR_LIT:GET>" , '<STR_LIT>' , re . compile ( "<STR_LIT>" ) ] , <EOL> ] <EOL> def __init__ ( self , config = None ) : <EOL> self . headers = { } <EOL> self . set_config ( config ) <EOL> self . git = Git ( "<STR_LIT>" ) <EOL> self . RE_SERVICES = [ ] <EOL> def set_config ( self , config ) : <EOL> self . config = config or { } <EOL> def set_config_setting ( self , key , value ) : <EOL> self . config [ key ] = value <EOL> def __call__ ( self , environ , start_response ) : <EOL> self . env = environ <EOL> body = self . call ( ) <EOL> start_response ( self . status , self . headers . items ( ) ) <EOL> return body <EOL> def call ( self ) : <EOL> match = self . match_routing ( self . env [ "<STR_LIT>" ] . lstrip ( '<STR_LIT:/>' ) , self . env [ "<STR_LIT>" ] ) <EOL> if not match : <EOL> return self . render_not_found ( ) <EOL> cmd , path , reqfile , rpc = match <EOL> self . rpc = rpc <EOL> self . reqfile = reqfile <EOL> if cmd == "<STR_LIT>" : <EOL> return self . render_method_not_allowed ( ) <EOL> self . dir = self . get_git_dir ( path ) <EOL> if not self . dir : <EOL> return self . render_not_found ( ) <EOL> func = getattr ( self , cmd ) <EOL> return func ( ) <EOL> def service_rpc ( self ) : <EOL> if not self . has_access ( self . rpc , True ) : <EOL> return self . render_no_access ( ) <EOL> input = self . read_body <EOL> git_cmd = "<STR_LIT>" if self . rpc == "<STR_LIT>" else "<STR_LIT>" <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = "<STR_LIT>" % self . rpc <EOL> return getattr ( self . git , git_cmd ) ( self . dir , { "<STR_LIT>" : input } , callback ) <EOL> def get_info_refs ( self ) : <EOL> service_name = self . get_service_type ( ) <EOL> if self . has_access ( service_name ) : <EOL> git_cmd = "<STR_LIT>" if service_name == "<STR_LIT>" else "<STR_LIT>" <EOL> refs = getattr ( self . git , git_cmd ) ( self . dir , { "<STR_LIT>" : True } ) <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = "<STR_LIT>" % service_name <EOL> self . hdr_nocache ( ) <EOL> def read_file ( ) : <EOL> yield self . pkt_write ( "<STR_LIT>" % service_name ) <EOL> yield self . pkt_flush <EOL> yield refs <EOL> return read_file ( ) <EOL> else : <EOL> return self . dumb_info_refs ( ) <EOL> def get_text_file ( self ) : <EOL> return self . send_file ( self . reqfile , "<STR_LIT>" ) <EOL> def dumb_info_refs ( self ) : <EOL> self . update_server_info ( ) <EOL> return self . send_file ( self . reqfile , "<STR_LIT>" ) <EOL> def get_info_packs ( self ) : <EOL> return self . send_file ( self . reqfile , "<STR_LIT>" ) <EOL> def get_loose_object ( self ) : <EOL> return self . send_file ( self . reqfile , "<STR_LIT>" , cached = True ) <EOL> def get_pack_file ( self ) : <EOL> return self . send_file ( self . reqfile , "<STR_LIT>" , cached = True ) <EOL> def get_idx_file ( self ) : <EOL> return self . send_file ( self . reqfile , "<STR_LIT>" , cached = True ) <EOL> def get_service_type ( self ) : <EOL> def get_param ( ) : <EOL> for query in self . env [ "<STR_LIT>" ] . split ( '<STR_LIT:&>' ) : <EOL> param = tuple ( query . split ( '<STR_LIT:=>' ) ) <EOL> if param and param [ <NUM_LIT:0> ] == "<STR_LIT>" : <EOL> return param [ <NUM_LIT:1> ] <EOL> service_type = get_param ( ) <EOL> if not service_type : <EOL> return False <EOL> if service_type [ <NUM_LIT:0> : <NUM_LIT:4> ] != '<STR_LIT>' : <EOL> return False <EOL> return service_type . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ classmethod <EOL> def match_routing ( cls , path_info , request_method ) : <EOL> for service in cls . SERVICES : <EOL> rpc = None <EOL> if len ( service ) == <NUM_LIT:4> : <EOL> method , handler , re_match , rpc = service <EOL> elif len ( service ) == <NUM_LIT:3> : <EOL> method , handler , re_match = service <EOL> m = re_match . match ( path_info ) <EOL> if m : <EOL> if method != request_method : <EOL> return [ "<STR_LIT>" , None , None , None ] <EOL> cmd = handler <EOL> path = m . group ( <NUM_LIT:1> ) <EOL> file = path_info . replace ( path + '<STR_LIT:/>' , '<STR_LIT>' ) <EOL> return [ cmd , path , file , rpc ] <EOL> return None <EOL> def send_file ( self , reqfile , content_type , cached = False ) : <EOL> reqfile = join ( self . dir , reqfile ) <EOL> if not self . is_subpath ( reqfile , self . dir ) : <EOL> return self . render_no_access ( ) <EOL> if not exists ( reqfile ) or not access ( reqfile , os . R_OK ) : <EOL> return self . render_not_found ( ) <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = content_type <EOL> self . headers [ "<STR_LIT>" ] = format_date_time ( getmtime ( reqfile ) ) <EOL> if cached : <EOL> self . hdr_cache_forenver ( ) <EOL> else : <EOL> self . hdr_nocache ( ) <EOL> size = getsize ( reqfile ) <EOL> if size : <EOL> self . headers [ "<STR_LIT>" ] = size <EOL> def read_file ( ) : <EOL> with open ( reqfile , "<STR_LIT:rb>" ) as f : <EOL> while True : <EOL> part = f . read ( <NUM_LIT> ) <EOL> if not part : <EOL> break <EOL> yield part <EOL> return read_file ( ) <EOL> else : <EOL> with open ( reqfile , "<STR_LIT:rb>" ) as f : <EOL> part = f . read ( ) <EOL> self . headers [ "<STR_LIT>" ] = str ( len ( part ) ) <EOL> return [ part ] <EOL> def update_server_info ( self ) : <EOL> self . git . update_server_info ( self . dir ) <EOL> @ property <EOL> def read_body ( self ) : <EOL> input = self . env [ "<STR_LIT>" ] <EOL> return input . read ( ) <EOL> @ property <EOL> def pkt_flush ( self ) : <EOL> return '<STR_LIT>' <EOL> def pkt_write ( self , str ) : <EOL> PKT_FORMAT = "<STR_LIT>" <EOL> return PKT_FORMAT . format ( len ( str ) + <NUM_LIT:4> , <EOL> str , <EOL> base = '<STR_LIT:x>' , <EOL> width = <NUM_LIT:4> , <EOL> fill = '<STR_LIT:0>' , <EOL> align = '<STR_LIT:>>' ) <EOL> def hdr_nocache ( self ) : <EOL> self . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> def hdr_cache_forenver ( self ) : <EOL> now = int ( time . time ( ) ) <EOL> self . headers [ "<STR_LIT>" ] = str ( now ) <EOL> self . headers [ "<STR_LIT>" ] = str ( now + <NUM_LIT> ) <EOL> self . headers [ "<STR_LIT>" ] = "<STR_LIT>" <EOL> def render_method_not_allowed ( self ) : <EOL> env = [ ] <EOL> if env [ "<STR_LIT>" ] == "<STR_LIT>" : <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = "<STR_LIT>" <EOL> return [ "<STR_LIT>" ] <EOL> else : <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = "<STR_LIT>" <EOL> return [ "<STR_LIT>" ] <EOL> def render_not_found ( self ) : <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = "<STR_LIT>" <EOL> return [ "<STR_LIT>" ] <EOL> def render_no_access ( self ) : <EOL> self . status = "<STR_LIT>" <EOL> self . headers [ "<STR_LIT:Content-Type>" ] = "<STR_LIT>" <EOL> return [ "<STR_LIT>" ] <EOL> def has_access ( self , rpc , check_content_type = False ) : <EOL> if check_content_type : <EOL> if self . env [ "<STR_LIT>" ] != "<STR_LIT>" % rpc : <EOL> return False <EOL> if rpc not in self . VALID_SERVICE_TYPES : <EOL> return False <EOL> if rpc == '<STR_LIT>' : <EOL> if "<STR_LIT>" in self . config : <EOL> return self . config . get ( "<STR_LIT>" ) <EOL> if rpc == '<STR_LIT>' : <EOL> if "<STR_LIT>" in self . config : <EOL> return self . config . get ( "<STR_LIT>" ) <EOL> return self . get_config_setting ( rpc ) <EOL> def get_config_setting ( self , service_name ) : <EOL> service_name = service_name . replace ( '<STR_LIT:->' , '<STR_LIT>' ) <EOL> setting = self . git . get_config_setting ( self . dir , "<STR_LIT>" % service_name ) <EOL> if service_name == '<STR_LIT>' : <EOL> return setting != '<STR_LIT:false>' <EOL> else : <EOL> return setting == '<STR_LIT:true>' <EOL> def get_git_dir ( self , path ) : <EOL> root = self . get_project_root ( ) <EOL> path = join ( root , path ) <EOL> if not self . is_subpath ( path , root ) : <EOL> return False <EOL> if exists ( path ) : <EOL> return path <EOL> return False <EOL> def get_project_root ( self ) : <EOL> root = self . config . get ( "<STR_LIT>" ) or os . getcwd ( ) <EOL> return root <EOL> def is_subpath ( self , path , checkpath ) : <EOL> path = unquote ( path ) <EOL> checkpath = unquote ( checkpath ) <EOL> checkpath = checkpath . replace ( "<STR_LIT>" , '<STR_LIT>' ) <EOL> if re . match ( "<STR_LIT>" % checkpath , path ) : <EOL> return True </s>
