<s> import BaseHTTPServer <EOL> import biplist <EOL> import re , hashlib <EOL> import time , base64 , random <EOL> import config <EOL> import fply <EOL> class DigestAuthHandler ( BaseHTTPServer . BaseHTTPRequestHandler ) : <EOL> realm = "<STR_LIT>" <EOL> def md5Join ( self , * args ) : <EOL> return hashlib . md5 ( "<STR_LIT::>" . join ( args ) ) . hexdigest ( ) <EOL> def requestAuth ( self ) : <EOL> now = str ( int ( time . time ( ) ) ) <EOL> randStr = "<STR_LIT>" . join ( chr ( int ( <NUM_LIT> * random . random ( ) ) ) for i in xrange ( <NUM_LIT:16> ) ) <EOL> self . nonce = base64 . b64encode ( ( "<STR_LIT>" % ( now , randStr ) ) ) <EOL> self . send_response ( <NUM_LIT> ) <EOL> self . send_header ( "<STR_LIT>" , <EOL> '<STR_LIT>' % ( self . realm , self . nonce ) ) <EOL> self . send_header ( "<STR_LIT>" , <NUM_LIT:0> ) <EOL> self . end_headers ( ) <EOL> def parseAuthHeader ( self ) : <EOL> auth = dict ( re . findall ( r'<STR_LIT>' , <EOL> self . headers . getheader ( "<STR_LIT>" ) or "<STR_LIT>" ) ) <EOL> if self . nonce != auth [ "<STR_LIT>" ] : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> hash_a1 = self . md5Join ( auth [ "<STR_LIT:username>" ] , self . realm , config . password ) <EOL> hash_a2 = self . md5Join ( self . command , auth [ "<STR_LIT>" ] ) <EOL> response = self . md5Join ( hash_a1 , self . nonce , hash_a2 ) <EOL> if response != auth [ "<STR_LIT>" ] : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def checkAuth ( self ) : <EOL> try : <EOL> if config . password is not None : <EOL> self . parseAuthHeader ( ) <EOL> return True <EOL> except : <EOL> self . requestAuth ( ) <EOL> return False <EOL> class AirPlayHandler ( DigestAuthHandler ) : <EOL> sys_version = "<STR_LIT>" <EOL> server_version = "<STR_LIT>" % ( config . server_name , config . server_version ) <EOL> def sendPList ( self , content , binary = False ) : <EOL> mime = "<STR_LIT>" if binary else "<STR_LIT>" <EOL> self . sendContent ( biplist . writePlistToString ( content , binary ) , mime ) <EOL> def sendContent ( self , content , contentType , X_Apple_ET = None ) : <EOL> self . send_response ( <NUM_LIT:200> ) <EOL> self . send_header ( "<STR_LIT:Content-Type>" , contentType ) <EOL> if X_Apple_ET is not None : <EOL> self . send_header ( "<STR_LIT>" , X_Apple_ET ) <EOL> self . send_header ( "<STR_LIT>" , len ( content ) ) <EOL> self . end_headers ( ) <EOL> self . wfile . write ( content ) <EOL> def parseToDict ( self , elements , delim = "<STR_LIT:=>" ) : <EOL> return dict ( ( ( k , v or None ) for k , _d , v in ( l . partition ( delim ) for l in elements if l ) ) ) <EOL> def readBody ( self ) : <EOL> length = int ( self . headers . getheader ( "<STR_LIT>" ) ) <EOL> return self . rfile . read ( length ) <EOL> def readPlist ( self ) : <EOL> return biplist . readPlistFromString ( self . readBody ( ) ) <EOL> def readSDP ( self ) : <EOL> return self . parseToDict ( self . readBody ( ) . split ( "<STR_LIT:\r\n>" ) ) <EOL> class FPLYHandler ( AirPlayHandler ) : <EOL> def setup ( self ) : <EOL> self . fply = fply . FPLY ( ) <EOL> AirPlayHandler . setup ( self ) <EOL> def do_POST ( self ) : <EOL> if self . path == "<STR_LIT>" : <EOL> self . fpSetup ( ) <EOL> def fpSetup ( self ) : <EOL> data = self . readBody ( ) <EOL> if len ( data ) == <NUM_LIT> : <EOL> answer = self . fply . phase1 ( data ) <EOL> else : <EOL> answer = self . fply . phase2 ( data ) <EOL> self . sendContent ( answer , "<STR_LIT>" ) </s>
