mBNj82ddlmZddlZddlmZddlmZddlm Z m Z ddl m Z m Z mZmZmZddlmZej&rdd lmZed ZGd d e ZGd de ZGddZGdde ZGdde ZGddeZGddZy)) annotationsN) getLogger)_typing)AsyncCooperativeLock AsyncLock)AsyncAuthManagerAsyncClientCertificateProviderClientCertificateexpiring_auth_has_expired ExpiringAuth)_TAuth) Neo4jErrorzneo4j.auth_managementc<eZdZUded<ddZddZ d dZy) AsyncStaticAuthManagerr_authc||_yNrselfauths m/Users/ahmed/devFolder/Ultron/claude-voice/.venv/lib/python3.12/site-packages/neo4j/_async/auth_management.py__init__zAsyncStaticAuthManager.__init__1  c"K|jSwrr)rs rget_authzAsyncStaticAuthManager.get_auth4szzs c Kyw)NF)rrerrors rhandle_security_exceptionz0AsyncStaticAuthManager.handle_security_exception7s N)rrreturnNoner$rrrr!rr$bool)__name__ __module__ __qualname____annotations__rrr"r rrrr.s. M#- rrcleZdZUded<ded<ded<ded< dd Zd Zdd Z dd Zy )AsyncNeo4jAuthTokenManagerzExpiringAuth | None _current_auth)t.Callable[[], t.Awaitable[ExpiringAuth]] _providerfrozenset[str]_handled_codesr_lockcL||_||_d|_t|_yr)r1r3r/rr4)rprovider handled_codess rrz#AsyncNeo4jAuthTokenManager.__init__Cs$ "+![ rcK |jd{|_|j t dy7!#t$r}tj d|d}~wwxYww)Nz.[ ] _: provider failed: %rzbAuth provider function passed to expiration_based AuthManager returned None, expected ExpiringAuth)r1r/ BaseExceptionlogr! TypeError)res r _refresh_authz(AsyncNeo4jAuthTokenManager._refresh_authMsi '+~~'7!7D     %C  & "8  IIF J  s1A#;9 ;A#; A AA  A#cZK|j4d{|j}| t|rBtj d|dnd|j d{|j}|J|j cdddd{S7|737 #1d{7swYyxYww)Nz*[ ] _: refreshing (%s)initztime out)r4r/r r:debugr=rrs rrz#AsyncNeo4jAuthTokenManager.get_authYs:::%%D|8> @"lF ((***))'''99::+:::sWB+BB+ABBB> B+ B B+BB+B(B B($B+cjK|j|jvry|j4d{|j}|G|j|k(r8t j d|j|jd{ dddd{y7k77 #1d{7swYyxYww)NFz0[ ] _: refreshing (error %s)T)coder3r4r/rr:r@r=)rrr!cur_auths rr"z4AsyncNeo4jAuthTokenManager.handle_security_exceptionfs ::T00 0:::))H# (= FJJ((***::+:::sW*B3BB3ABBB B3BB3BB3B0$B' %B0,B3N)r6r0r7r2r$r%r&r')r)r*r+r,rr=rr"r rrr.r.=sb&&88"" !;!&!  !    #-   rr.cVeZdZdZeddZe ddZe ddZy) AsyncAuthManagersa A collection of :class:`.AsyncAuthManager` factories. .. versionadded:: 5.8 .. versionchanged:: 5.12 * Method ``expiration_based()`` was renamed to :meth:`bearer`. * Added :meth:`basic`. .. versionchanged:: 5.14 Stabilized from preview. ct|S)a Create a static auth manager. The manager will always return the auth info provided at its creation. Example:: # NOTE: this example is for illustration purposes only. # The driver will automatically wrap static auth info in a # static auth manager. import neo4j from neo4j.auth_management import AsyncAuthManagers auth = neo4j.basic_auth("neo4j", "password") with neo4j.GraphDatabase.driver( "neo4j://example.com:7687", auth=AsyncAuthManagers.static(auth) # auth=auth # this is equivalent ) as driver: ... # do stuff :param auth: The auth to return. :returns: An instance of an implementation of :class:`.AsyncAuthManager` that always returns the same auth. .. versionadded:: 5.8 .. versionchanged:: 5.14 Stabilized from preview. )r)rs rstaticzAsyncAuthManagers.staticsH&d++rc>td}dfd }t||S)a, Create an auth manager handling basic auth password rotation. This factory wraps the provider function in an auth manager implementation that caches the provided auth info until the server notifies the driver that the auth info has expired (by returning an error that indicates that the password is invalid). Note that this implies that the provider function will be called again if it provides wrong auth info, potentially deferring failure due to a wrong password or username. .. warning:: The provider function **must not** interact with the driver in any way as this can cause deadlocks and undefined behaviour. The provider function must only ever return auth information belonging to the same identity. Switching identities is undefined behavior. You may use :ref:`session-level authentication` for such use-cases. Example:: import neo4j from neo4j.auth_management import ( AsyncAuthManagers, ExpiringAuth, ) async def auth_provider(): # some way of getting a token user, password = await get_current_auth() return (user, password) with neo4j.GraphDatabase.driver( "neo4j://example.com:7687", auth=AsyncAuthManagers.basic(auth_provider) ) as driver: ... # do stuff :param provider: A callable that provides new auth info whenever the server notifies the driver that the previous auth info is invalid. :returns: An instance of an implementation of :class:`.AsyncAuthManager` that returns auth info from the given provider and refreshes it, calling the provider again, when the auth info was rejected by the server. .. versionadded:: 5.12 .. versionchanged:: 5.14 Stabilized from preview. )%Neo.ClientError.Security.Unauthorizedc@Ktd{S7wr)r )r6srwrapped_providerz1AsyncAuthManagers.basic..wrapped_providershj 01 1 0s  )r$r  frozensetr.)r6r7rKs` rbasiczAsyncAuthManagers.basics&z""LM  2**:MJJrc0td}t||S)a Create an auth manager for potentially expiring bearer auth tokens. This factory wraps the provider function in an auth manager implementation that caches the provided auth info until either the :attr:`.ExpiringAuth.expires_at` exceeded or the server notified the driver that the auth info has expired (by returning an error that indicates that the bearer auth token has expired). .. warning:: The provider function **must not** interact with the driver in any way as this can cause deadlocks and undefined behaviour. The provider function must only ever return auth information belonging to the same identity. Switching identities is undefined behavior. You may use :ref:`session-level authentication` for such use-cases. Example:: import neo4j from neo4j.auth_management import ( AsyncAuthManagers, ExpiringAuth, ) async def auth_provider(): # some way of getting a token sso_token = await get_sso_token() # assume we know our tokens expire every 60 seconds expires_in = 60 # Include a little buffer so that we fetch a new token # *before* the old one expires expires_in -= 10 auth = neo4j.bearer_auth(sso_token) return ExpiringAuth(auth=auth).expires_in(expires_in) with neo4j.GraphDatabase.driver( "neo4j://example.com:7687", auth=AsyncAuthManagers.bearer(auth_provider) ) as driver: ... # do stuff :param provider: A callable that provides a :class:`.ExpiringAuth` instance. :returns: An instance of an implementation of :class:`.AsyncAuthManager` that returns auth info from the given provider and refreshes it, calling the provider again, when the auth info expires (either because it's reached its expiry time or because the server flagged it as expired). .. versionadded:: 5.12 .. versionchanged:: 5.14 Stabilized from preview. )z%Neo.ClientError.Security.TokenExpiredrIrL)r6r7s rbearerzAsyncAuthManagers.bearers#F"  *(MBBrN)rrr$r )r6z#t.Callable[[], t.Awaitable[_TAuth]]r$r )r6r0r$r )r)r*r+__doc__ staticmethodrGrNrPr rrrErEvsu #,#,JAK5AK AKAKFHC;HC HCHCrrEc(eZdZUded<ddZddZy)%_AsyncStaticClientCertificateProviderClientCertificate | None_certc||_yrrVrcerts rrz._AsyncStaticClientCertificateProvider.__init__=rrc6K|jdc}|_|SwrrXrYs rget_certificatez5_AsyncStaticClientCertificateProvider.get_certificate@s::tdj sNrZr r$r%r$rU)r)r*r+r,rr\r rrrTrT:s ##rrTc6eZdZdZej ddZy)&AsyncRotatingClientCertificateProvidera[ Abstract base class for certificate providers that can rotate certificates. The provider will make the driver use the initial certificate for all connections until the certificate is updated using the :meth:`update_certificate` method. From that point on, the new certificate will be used for all new connections until :meth:`update_certificate` is called again and so on. Example:: from neo4j import AsyncGraphDatabase from neo4j.auth_management import ( ClientCertificate, AsyncClientCertificateProviders, ) provider = AsyncClientCertificateProviders.rotating( ClientCertificate( certfile="path/to/certfile.pem", keyfile="path/to/keyfile.pem", password=lambda: "super_secret_password" ) ) driver = AsyncGraphDatabase.driver( # secure driver must be configured for client certificate # to be used: (...+s[sc] scheme or encrypted=True) "neo4j+s://example.com:7687", # auth still required as before, unless server is configured to not # use authentication auth=("neo4j", "password"), client_certificate=provider ) # do work with the driver, until the certificate needs to be rotated ... await provider.update_certificate( ClientCertificate( certfile="path/to/new/certfile.pem", keyfile="path/to/new/keyfile.pem", password=lambda: "new_super_secret_password" ) ) # do more work with the driver, until the certificate needs to be # rotated again ... .. versionadded:: 5.19 .. versionchanged:: 5.24 Turned this class into an abstract class to make the actual implementation internal. This entails removing the possibility to directly instantiate this class. Please use the factory method :meth:`.AsyncClientCertificateProviders.rotating` instead. .. versionchanged:: 5.27 Stabilized from preview. c Kyw)z2Update the certificate to use for new connections.Nr rYs rupdate_certificatez9AsyncRotatingClientCertificateProvider.update_certificatesr#Nr])r)r*r+rQabcabstractmethodrbr rrr`r`Es$<| AArr`c$eZdZddZddZddZy),_AsyncNeo4jRotatingClientCertificateProviderc0||_t|_yr)rVrr4)r initial_certs rrz5_AsyncNeo4jRotatingClientCertificateProvider.__init__s/; )+ rcK|j4d{|jdc}|_|cdddd{S7+7#1d{7swYyxYwwrr4rVrYs rr\z<_AsyncNeo4jRotatingClientCertificateProvider.get_certificates7:::#zz4 D$*::::::s@A?AA AAAAA A AAcK|j4d{||_dddd{y77#1d{7swYyxYwwrrjrYs rrbz?_AsyncNeo4jRotatingClientCertificateProvider.update_certificates):::DJ::::::s9A 0A 4 A 2A A A= AA N)rhr r$r%r^r])r)r*r+rr\rbr rrrfrfs, rrfc<eZdZdZeddZe ddZy)AsyncClientCertificateProvidersz A collection of :class:`.AsyncClientCertificateProvider` factories. .. versionadded:: 5.19 .. versionchanged:: 5.27 Stabilized from preview. ct|S)z Create a static client certificate provider. The provider simply makes the driver use the given certificate for all connections. )rT)rZs rrGz&AsyncClientCertificateProviders.statics5T::rct|S)z Create certificate provider that allows for rotating certificates. .. seealso:: :class:`.AsyncRotatingClientCertificateProvider` )rf)rhs rrotatingz(AsyncClientCertificateProviders.rotatingsrzs"# ??''( - 6!16rACACH,JAA-KAAH*"JJr