o
    ga                     @   s  d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dl	Z	d dl
Z
d dlmZ d dlmZ d dlmZ zd dlZW n eyG   dZY nw ejeZejedZejedZG dd dejjZG d	d
 d
ejZG dd dZG dd dejjZG dd dejjZ G dd de	j!Z"e#dG dd de	j!Z$dd Z%G dd de	j!Z&da'dd Z(dd Z)e*dkre	+  dS dS )    N)hashlib_helper)threading_helper)warnings_helperzkeycert.pemzkeycert2.pemc                   @   s    e Zd ZdZdd Zdd ZdS )LoopbackHttpServerz^HTTP server w/ a few modifications that make it useful for
    loopback testing purposes.
    c                 C   s"   t jj| || | jd d S )Ng?)httpserver
HTTPServer__init__socket
settimeout)selfZserver_addressZRequestHandlerClass r   P/Users/admin/.pyenv/versions/3.10.0/lib/python3.10/test/test_urllib2_localnet.pyr	   "   s
   
zLoopbackHttpServer.__init__c                 C   s    | j  \}}|d ||fS )zHTTPServer method, overridden.g      $@N)r
   acceptr   )r   requestZclient_addressr   r   r   get_request+   s   
zLoopbackHttpServer.get_requestN)__name__
__module____qualname____doc__r	   r   r   r   r   r   r      s    	r   c                   @   (   e Zd ZdZdd Zdd Zdd ZdS )	LoopbackHttpServerThreadz2Stoppable thread that runs a loopback http server.c                 C   s<   t j|  d| _t  | _d|_td|| _| jj	| _
d S )NFzHTTP/1.0)z	127.0.0.1r   )	threadingThreadr	   _stop_serverEventreadyZprotocol_versionr   httpdZserver_portportr   request_handlerr   r   r   r	   :   s   
z!LoopbackHttpServerThread.__init__c                 C   s   d| _ |   | j  dS )z.Stops the webserver if it's currently running.TN)r   joinr   Zserver_closer   r   r   r   stopC   s   zLoopbackHttpServerThread.stopc                 C   s(   | j   | js| j  | jrd S d S N)r   setr   r   handle_requestr"   r   r   r   runK   s   

zLoopbackHttpServerThread.runN)r   r   r   r   r	   r#   r'   r   r   r   r   r   7   s
    	r   c                   @   sX   e Zd ZdZdd Zdd Zdd Zdd	 Zd
d Zdd Z	dd Z
dd Zdd ZdS )DigestAuthHandlerz-Handler for performing digest authentication.c                 C   s"   d| _ g | _i | _d| _d| _d S )Nr   z
Test Realmauth)_request_num_nonces_users_realm_name_qopr"   r   r   r   r	   U   s
   
zDigestAuthHandler.__init__c                 C   
   || _ d S r$   )r.   )r   Zqopr   r   r   set_qop\      
zDigestAuthHandler.set_qopc                 C   s   t |tsJ || _d S r$   )
isinstancedictr,   )r   Zusersr   r   r   	set_users_   s   
zDigestAuthHandler.set_usersc                 C   r/   r$   )r-   )r   Zrealmr   r   r   	set_realmc   r1   zDigestAuthHandler.set_realmc                 C   s8   |  j d7  _ tt| j d }| j| |S )N   ascii)r*   hashlibmd5strencode	hexdigestr+   append)r   noncer   r   r   _generate_noncef   s   z!DigestAuthHandler._generate_noncec                 C   s   | d}||d d  }|d}i }|D ](}|d\}}| }|d dkr6|d dkr6|dd }n| }|||< q|S )N r6   ,=r   ")findsplitstrip)r   Zauth_strZfirst_space_indexparts	auth_dictpartnamevaluer   r   r   _create_auth_dictl   s   


z#DigestAuthHandler._create_auth_dictc                 C   s   i }| | ||d< ||d< ||d< d| }t|d }d| }t|d }	||d< |	|d< d	| }
t|
d }||d
 kS )Npasswordmethoduriz#%(username)s:%(realm)s:%(password)sr7   z%(method)s:%(uri)sHA1HA2z3%(HA1)s:%(nonce)s:%(nc)s:%(cnonce)s:%(qop)s:%(HA2)sresponse)updater8   r9   r;   r<   )r   rI   rN   rO   rP   Z
final_dictZHA1_strrQ   ZHA2_strrR   Zresponse_strrS   r   r   r   _validate_auth}   s    
z DigestAuthHandler._validate_authc                 C   sN   | dd |dd |dd| j| j|  f  |  |jd dS )	Ni  zProxy Authentication RequiredContent-Type	text/htmlzProxy-Authenticatez(Digest realm="%s", qop="%s",nonce="%s", s   Proxy Authentication Required.F)send_responsesend_headerr-   r.   r?   end_headerswfilewriter   r   r   r   _return_auth_challenge   s   z(DigestAuthHandler._return_auth_challengec                 C   s   t | jdkr	dS d|jvr| |S | |jd }|d | jv r*| j|d  }n| |S |d| jvr<| |S | j|d  d}|j|j	fD ]}| 
|||j|rYd}qL|sa| |S dS )a  Performs digest authentication on the given HTTP request
        handler.  Returns True if authentication was successful, False
        otherwise.

        If no users have been set, then digest auth is effectively
        disabled and this method will always return True.
        r   TzProxy-Authorizationusernamer>   FN)lenr,   headersr]   rM   getr+   removepath
short_pathrU   Zcommand)r   r    rI   rN   Zauth_validatedrc   r   r   r   r&      s2   	




z DigestAuthHandler.handle_requestN)r   r   r   r   r	   r0   r4   r5   r?   rM   rU   r]   r&   r   r   r   r   r(   R   s    r(   c                   @   sf   e Zd ZdZdZdZdZdeef Ze	e
ddZdd Zd	d
 Zdd Zdd Zdd ZdS )BasicAuthHandlerz,Handler for performing basic authentication.testUsertestPassTestz%s:%sr7   c                 O   s    t jjj| g|R i | d S r$   )r   r   BaseHTTPRequestHandlerr	   )r   argskwargsr   r   r   r	      s    zBasicAuthHandler.__init__c                 G      d S r$   r   r   formatrj   r   r   r   log_message   s   zBasicAuthHandler.log_messagec                 C   s"   |  d | dd |   d S )N   Content-typerW   )rX   rY   rZ   r"   r   r   r   do_HEAD   s   
zBasicAuthHandler.do_HEADc                 C   s4   |  d | dd| j  | dd |   d S )Ni  zWWW-AuthenticatezBasic realm="%s"rq   rW   )rX   rY   REALMrZ   r"   r   r   r   do_AUTHHEAD   s   
zBasicAuthHandler.do_AUTHHEADc                 C   sl   | j dds|   | jd d S | j ddd| j kr0| d |   | jd d S |   d S )NZAuthorization s   No Auth header receivedzBasic rp   s   It works)r`   ra   rt   r[   r\   ENCODED_AUTHrX   rZ   r"   r   r   r   do_GET   s   
zBasicAuthHandler.do_GETN)r   r   r   r   USERPASSWDrs   ZUSER_PASSWDbase64Z	b64encoder;   decoderv   r	   ro   rr   rt   rw   r   r   r   r   re      s    re   c                   @   r   )	FakeProxyHandlerzThis is a 'fake proxy' that makes it look like the entire
    internet has gone down due to a sudden zombie invasion.  It main
    utility is in providing us with authentication support for
    testing.
    c                 O   s&   || _ tjjj| g|R i | d S r$   )digest_auth_handlerr   r   ri   r	   )r   r}   rj   rk   r   r   r   r	      s    zFakeProxyHandler.__init__c                 G   rl   r$   r   rm   r   r   r   ro     s   zFakeProxyHandler.log_messagec                 C   sz   t j| jd\}}}}}}|| _| j| r;| dd | dd | 	  | j
td| j d | j
d d S d S )	Nr   rp   ZOKrV   rW   zYou've reached %s!<BR>r7   sF   Our apologies, but our server is down due to a sudden zombie invasion.)urllibparseurlparserc   rd   r}   r&   rX   rY   rZ   r[   r\   bytes)r   Zscmnetlocrc   paramsqueryfragmentr   r   r   rw   	  s   zFakeProxyHandler.do_GETN)r   r   r   r   r	   ro   rw   r   r   r   r   r|      s
    r|   c                       sP   e Zd ZdZdZdZdZ fddZdd Z fd	d
Z	dd Z
dd Z  ZS )BasicAuthTestsrf   rg   Z	Incorrectrh   c                    sT   t t|   dd }t|| _| | j d| jj | _| j	  | jj
  d S )Nc                  _   s   t | i |S r$   )re   rj   rk   r   r   r   #http_server_with_basic_auth_handler!  s   zABasicAuthTests.setUp.<locals>.http_server_with_basic_auth_handlerzhttp://127.0.0.1:%s)superr   setUpr   r   
addCleanupstop_serverr   
server_urlstartr   wait)r   r   	__class__r   r   r     s   

zBasicAuthTests.setUpc                 C      | j   d | _ d S r$   r   r#   r"   r   r   r   r   )     

zBasicAuthTests.stop_serverc                    s   t t|   d S r$   )r   r   tearDownr"   r   r   r   r   -  s   zBasicAuthTests.tearDownc                 C   sz   t j }|| j| j| j| j t jt j	| z| 
t j| j W d S  t jjy<   | d| j  Y d S w )Nz!Basic auth failed for the url: %s)r~   r   HTTPBasicAuthHandleradd_passwordrs   r   rx   ry   install_openerbuild_opener
assertTrueurlopenerror	HTTPErrorfailr   Zahr   r   r   test_basic_auth_success0  s   
z&BasicAuthTests.test_basic_auth_successc                 C   sR   t j }|| j| j| j| j t jt j	| | 
t jjt jj| j d S r$   )r~   r   r   r   rs   r   rx   INCORRECT_PASSWDr   r   assertRaisesr   r   r   r   r   r   r   test_basic_auth_httperror9  s   
z(BasicAuthTests.test_basic_auth_httperror)r   r   r   rx   ry   r   rs   r   r   r   r   r   __classcell__r   r   r   r   r     s    	r   r9   c                       sT   e Zd ZdZdZdZdZ fddZdd Zd	d
 Z	dd Z
dd Zdd Z  ZS )ProxyAuthTestszhttp://localhostZtesterZtest123Z	TestRealmc                    s   t t   dd } |tj  dtjd< dtjd< t  _ j	 j
 ji  j j  fdd}t| _  j  j   jj  d jj }tjd	|i}tj  _tj| j _d S )
Nc                 S      t j  t j|  d S r$   osenvironclearrT   Zold_environr   r   r   restore_environK     
z-ProxyAuthTests.setUp.<locals>.restore_environru   NO_PROXYno_proxyc                     s   t  jg| R i |S r$   )r|   r}   r   r"   r   r   create_fake_proxy_handlerV  s   z7ProxyAuthTests.setUp.<locals>.create_fake_proxy_handlerzhttp://127.0.0.1:%dr   )r   r   r   r   r   r   copyr(   r}   r4   rx   ry   r5   rs   r   r   r   r   r   r   r   r~   r   ZProxyHandlerZProxyDigestAuthHandlerproxy_digest_handlerr   opener)r   r   r   Z	proxy_urlhandlerr   r"   r   r   H  s&   




zProxyAuthTests.setUpc                 C   r   r$   r   r"   r   r   r   r   c  r   zProxyAuthTests.stop_serverc                 C   sF   | j | j| j| j| jd  | jd | t	j
j| jj| j d S )NZbadr)   )r   r   rs   URLrx   ry   r}   r0   r   r~   r   r   r   openr"   r   r   r   -test_proxy_with_bad_password_raises_httperrorg  s   
z<ProxyAuthTests.test_proxy_with_bad_password_raises_httperrorc                 C   s(   | j d | tjj| jj| j d S Nr)   )	r}   r0   r   r~   r   r   r   r   r   r"   r   r   r   ,test_proxy_with_no_password_raises_httperroro  s
   
z;ProxyAuthTests.test_proxy_with_no_password_raises_httperrorc                 C   s|   | j | j| j| j| j | jd | j	| j}|
 r,	 |
 sW d    d S W d    d S 1 s7w   Y  d S r   )r   r   rs   r   rx   ry   r}   r0   r   r   readr   resultr   r   r   test_proxy_qop_auth_worksu  s   
"z(ProxyAuthTests.test_proxy_qop_auth_worksc                 C   s   | j | j| j| j| j | jd z	| j	| j}W n t
jjy(   Y d S w | | r=	 | s0W d    d S W d    d S 1 sHw   Y  d S )Nzauth-int)r   r   rs   r   rx   ry   r}   r0   r   r   r~   r   URLErrorr   r   r   r   r   0test_proxy_qop_auth_int_works_or_throws_urlerror}  s    
"z?ProxyAuthTests.test_proxy_qop_auth_int_works_or_throws_urlerror)r   r   r   r   rx   ry   rs   r   r   r   r   r   r   r   r   r   r   r   r   @  s    r   c                    s   G  fdddt jj  S )Nc                       sB   e Zd ZdZg Zg ZdZdd Zdd Z fddZ	d	d
 Z
dS )z1GetRequestHandler.<locals>.FakeHTTPRequestHandlerz	TestHTTP/P   c                 S   s0   |   }|r| j|}||d  }|sd S d S r$   )	send_headr[   r\   )r   bodydoner   r   r   rw     s
   z8GetRequestHandler.<locals>.FakeHTTPRequestHandler.do_GETc                 S   s2   | j d }| jt|}|   | j| d S )NzContent-Length)r`   Zrfiler   intrw   requestsr=   )r   Zcontent_lengthZ	post_datar   r   r   do_POST  s   
z9GetRequestHandler.<locals>.FakeHTTPRequestHandler.do_POSTc                    s|   | j  _| j| j d\}}}| | |D ]\}}| ||d| ji  q|r8| dd | 	  |S | 	  d S )Nr   r   rq   z
text/plain)
r`   headers_receivedr   r=   rc   poprX   rY   r   rZ   )r   Zresponse_coder`   r   headerrL   FakeHTTPRequestHandler	responsesr   r   r     s   
z;GetRequestHandler.<locals>.FakeHTTPRequestHandler.send_headc                 W   rl   r$   r   )r   rj   r   r   r   ro     s   z=GetRequestHandler.<locals>.FakeHTTPRequestHandler.log_messageN)r   r   r   Zserver_versionr   r   r   rw   r   r   ro   r   r   r   r   r     s    r   )r   r   ri   )r   r   r   r   GetRequestHandler  s   &r   c                       s   e Zd ZdZ fddZd+ddZdd Zd+d	d
Zd+ddZdd Z	dd Z
dd Zdd Zdd Zdd Zdd Zdd Zdd Zdd  Zd!d" Zd#d$ Zd%d& Zd'd( Zd)d* Z  ZS ),TestUrlopenaN  Tests urllib.request.urlopen using the network.

    These tests are not exhaustive.  Assuming that testing using files does a
    good job overall of some of the basic interface features.  There are no
    tests exercising the optional 'data' and 'proxies' arguments.  No tests
    for transparent redirection have been written.
    c                    sN   t t|   | tjj dd }| |tj	  dtjd< dtjd< d S )Nc                 S   r   r$   r   r   r   r   r   r     r   z*TestUrlopen.setUp.<locals>.restore_environ*r   r   )
r   r   r   r   r~   r   
urlcleanupr   r   r   )r   r   r   r   r   r     s   
zTestUrlopen.setUpNc                 K   sx   g }t jj||fi |}z$||d ||  ||d ||  W |  n|  w d	|S )Nrp   i       )
r~   r   r   extend	readlinesr=   readliner   closer!   )r   urldatark   lfr   r   r   r     s   
zTestUrlopen.urlopenc                 C   r   r$   r   r"   r   r   r   r     r   zTestUrlopen.stop_serverc                 C   sZ   |d u r
dg dfg}t |}t|| _| | j | j  | jj  | jj}||_|S )Nrp   s   we don't care)	r   r   r   r   r   r   r   r   r   )r   r   r   r   r   r   r   start_server  s   

zTestUrlopen.start_serverc                 K   s^   t tjds| d ddlm} |d u rdg dfg}t|}|| fd|i|}|j|_|S )NZHTTPSHandlerzssl support requiredr   )make_https_serverrp      we care a bitZhandler_class)hasattrr~   r   skipTestZtest.ssl_serversr   r   r   )r   r   rk   r   r   r   r   r   r   start_https_server  s   
zTestUrlopen.start_https_serverc                 C   sV   d}ddgdfdg |fg}|  |}| d|j }| || | |jddg d S )	Ns   We got here...i.  )ZLocationz(http://localhost:%(port)s/somewhere_elseru   rp   http://localhost:%s//z/somewhere_elser   r   r   assertEqualr   )r   expected_responser   r   r   r   r   r   test_redirection  s   
zTestUrlopen.test_redirectionc                 C   s@   d}d}ddg|fg}|  |}| d|j }| || d S )Ns   hello worlds   a
hello worl
1
d
0
rp   )zTransfer-EncodingZchunkedr   )r   r   r   r   )r   r   Zchunked_startrS   r   r   r   r   r   test_chunked  s   
zTestUrlopen.test_chunkedc              
   C   s   d}|  dg |fg}z
| d|j  W n tjjy1 } z| }|  W Y d }~n
d }~ww | d | 	|| | 	|j
dg d S )Ns   Bad bad bad...i  zhttp://localhost:%s/weeblez404 should raise URLErrorz/weeble)r   r   r   r~   r   r   r   r   r   r   r   )r   r   r   r   r   r   r   r   test_404  s   
zTestUrlopen.test_404c                 C   sF   d}|  dg |fg}| d|j }| || | |jdg d S )N   pycon 2008...rp   http://localhost:%s/bizarre/bizarrer   r   r   r   r   r   r   r   test_200#  s
   zTestUrlopen.test_200c                 C   sJ   d}|  dg |fg}| d|j d}| || | |jddg d S )Nr   rp   r   s   get=with_feelingr   r   r   r   r   r   test_200_with_parameters*  s   z$TestUrlopen.test_200_with_parametersc                 C   s8   |   }tjtd}| jd|j |d}| |d d S )NZcafilehttps://localhost:%s/bizarrecontextr   )r   sslcreate_default_contextCERT_localhostr   r   r   )r   r   r   r   r   r   r   
test_https2  s   zTestUrlopen.test_httpsc              	   C   s   | j td}tdtfi | jd|j td}| |d | t	j
j}| jd|j td W d    n1 s:w   Y  | j td}| t	j
j}| jd|j td W d    n1 saw   Y  W d    d S W d    d S 1 syw   Y  d S )Ncertfileru   r   r   r   )r   r   r   check_warningsDeprecationWarningr   r   r   r   r~   r   r   CERT_fakehostname)r   r   r   cmr   r   r   test_https_with_cafile8  s(   "z"TestUrlopen.test_https_with_cafilec              	   C   s   | j td}tdtf2 | tjj}| j	d|j
 dd W d    n1 s*w   Y  W d    d S W d    d S 1 sBw   Y  d S )Nr   ru   r   T)Z	cadefault)r   r   r   r   r   r   r~   r   r   r   r   )r   r   r   r   r   r   test_https_with_cadefaultI  s   "z%TestUrlopen.test_https_with_cadefaultc                    s   t d u r	| d t js| d d   fdd}t t j}|| | j|td}t jtd}| j	d|j
 |d |  d	 d S )
Nzssl module requiredzSNI support required in OpenSSLc                    s   | d S r$   r   )Zssl_sockZserver_nameZinitial_contextZsni_namer   r   cb_sniW  s   z*TestUrlopen.test_https_sni.<locals>.cb_sni)r   r   r   zhttps://localhost:%sr   	localhost)r   r   ZHAS_SNIZ
SSLContextZPROTOCOL_TLS_SERVERZset_servername_callbackr   r   r   r   r   r   )r   r   r   r   r   r   r   test_https_sniQ  s   


zTestUrlopen.test_https_snic                 C   sd   |   }tjjd|j ddid}tj| W d    n1 s"w   Y  | |jd d d S )Nr   ZRangezbytes=20-39)r`   )r   r~   r   ZRequestr   r   r   r   )r   r   Zreqr   r   r   test_sending_headersa  s   z TestUrlopen.test_sending_headersc                 C   sp   |   }tjd|j !}dD ]}| t||d|  q| | d W d    d S 1 s1w   Y  d S )Nhttp://localhost:%s)r   r   infogeturlz3object returned from urlopen lacks the %s attributezcalling 'read' failed)r   r~   r   r   r   r   r   r   )r   r   open_urlattrr   r   r   
test_basici  s   "zTestUrlopen.test_basicc                 C   sl   |   }tjd|j }| | }W d    n1 sw   Y  | |tjj	d | 
| d d S )Nr  zEobject returned by 'info' is not an instance of email.message.MessageZplain)r   r~   r   r   r   r  ZassertIsInstanceemailmessageZMessager   Zget_content_subtype)r   r   r  Zinfo_objr   r   r   	test_infoq  s   
zTestUrlopen.test_infoc                 C   s\   |   }tjd|j }| | }W d    n1 sw   Y  | |d|j  d S )Nr  )r   r~   r   r   r   r  r   )r   r   r  r   r   r   r   test_geturl|  s   
zTestUrlopen.test_geturlc                 C   sB   d}|  dg |fg}tjd|j }|D ]}| || qd S )Nr   rp   r  )r   r~   r   r   r   r   )r   r   r   r   liner   r   r   test_iteration  s   zTestUrlopen.test_iterationc              
   C   s   ddddd d g}d |}| dg |fg}tjd	|j }t|D ]\}}| ||| d
|t|| t|f  q%| |d t| d S )Ns   We
s   got
s   here
s	   verylong i       
r   rp   r  zQFetched line number %s doesn't match expected:
    Expected length was %s, got %sr6   )	r!   r   r~   r   r   r   	enumerater   r_   )r   linesr   r   r   indexr  r   r   r   test_line_iteration  s   
zTestUrlopen.test_line_iterationr$   )r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r  r  r  r  r  r  r  r   r   r   r   r   r     s,    


r   c                   C   s   t  ad S r$   )r   Zthreading_setupthreads_keyr   r   r   r   setUpModule  s   r  c                   C   s   t r	tjt   d S d S r$   )r  r   Zthreading_cleanupr   r   r   r   tearDownModule  s   r  __main__),rz   r   r	  urllib.parser~   Zurllib.requestZhttp.serverr   r   Zunittestr8   Ztest.supportr   r   r   r   ImportErrorrc   dirname__file__herer!   r   r   r   r   r   r   r   r(   ri   re   r|   ZTestCaser   Zrequires_hashdigestr   r   r   r  r  r  r   mainr   r   r   r   <module>   sJ    y,!(M+ `