o
    H\jd                     @  s*  d Z ddlmZ ddlZddlZddlmZmZmZ ddlm	Z	 e	e
 jd Zejdee ejdZedurEeedsEejd= dd	lmZ dd
lmZmZ ddlmZ ddlmZ ddlmZ ddl m!Z! ddl"m#Z# ddl$m%Z%m&Z& dddZ'G dd dej(Z)e*dkrej+dd dS dS )z:Standalone Phase 1 site, key lifecycle, and replay checks.    )annotationsN)datetime	timedeltatimezone)Path   platform__path__)rsa)EncodingPublicFormat)create_engine)Session)
StaticPool)Base)registry)PlanTenantreturnstrc                  C  s(   t jddd} |  tjtjdS )Ni  i   )public_exponentkey_sizezutf-8)	r
   generate_private_key
public_keypublic_bytesr   PEMr   SubjectPublicKeyInfodecode)key r   &tests/platform/site_registry_checks.pypublic_key_pem   s   r!   c                   @  s<   e Zd Zdd Zdd Zdd Zdd Zd	d
 Zdd ZdS )SiteRegistryChecksc                   s   t ddditdd _tj j tj _ fddt_t	 jdd$}|
td	d	d
d |
tdddddd	d |  W d    d S 1 sKw   Y  d S )Nz	sqlite://check_same_threadFT)connect_args	poolclassfuturec                     s   t  jddS )NTr&   )r   enginer   selfr   r    <lambda>.   s    z*SiteRegistryChecks.setUp.<locals>.<lambda>r'   testTest)plan_idcodenametenant-azTenant Aactive)	tenant_idslugr0   status	is_activer.   )r   r   r(   r   metadata
create_allr   get_control_plane_sessionoriginal_session_factoryr   addr   r   commit)r*   sessionr   r)   r    setUp%   s.   

"zSiteRegistryChecks.setUpc                 C  s   | j t_| j  d S )N)r:   r   r9   r(   disposer)   r   r   r    tearDown=   s   zSiteRegistryChecks.tearDownc                 C  s   t jdddddt d d S )Nsite-ar1   zhttps://example.testzabsolutems-apikey-1)site_idr3   issueraudiencekey_idr!   )r   register_bootstrap_siter!   r)   r   r   r    registerA   s   
zSiteRegistryChecks.registerc                 C  s6   |    td}| d|d  | d|d  d S )NrB   rA   rC   r1   r3   )rH   r   get_active_signing_keyassertEqual)r*   r   r   r   r    &test_bootstrap_registration_and_lookupK   s   
z9SiteRegistryChecks.test_bootstrap_registration_and_lookupc                 C  sp   |    ttjtdd }tjdd|d | t	d tjdd|d W d    d S 1 s1w   Y  d S )N   )minutesrA   z
same-token)rC   jti
expires_atreplay)
rH   r   nowr   utcr   r   consume_jtiassertRaisesRegexPermissionError)r*   expiryr   r   r    test_replayed_jti_is_rejectedQ   s   "z0SiteRegistryChecks.test_replayed_jti_is_rejectedc                 C  s`   |    tjdddt d | td | td tjdddd | td d S )NrA   rB   zkey-2)rC   current_key_id
new_key_idr!   )rC   rF   rX   )rH   r   rotate_signing_keyr!   assertIsNotNonerI   revoke_signing_keyassertIsNoner)   r   r   r    test_rotation_and_revocationX   s   z/SiteRegistryChecks.test_rotation_and_revocationN)	__name__
__module____qualname__r>   r@   rH   rK   rW   r^   r   r   r   r    r"   $   s    
r"   __main__)	verbosity)r   r   ),__doc__
__future__r   sysunittestr   r   r   pathlibr   __file__resolveparentsPROJECT_ROOTpathinsertr   modulesgetloaded_platformhasattr)cryptography.hazmat.primitives.asymmetricr
   ,cryptography.hazmat.primitives.serializationr   r   
sqlalchemyr   sqlalchemy.ormr   sqlalchemy.poolr   	config.dbr   platform.sitesr   platform.tenants.modelsr   r   r!   TestCaser"   r_   mainr   r   r   r    <module>   s0    
F