tessl/pypi-neo4j

Neo4j Bolt driver for Python providing database connectivity and query execution

—

Pending

Overview

Eval results

Files

Authentication

Name: tessl/pypi-neo4j
Author: tessl

Authentication system supporting multiple authentication schemes including basic, Kerberos, bearer tokens, and custom authentication mechanisms. The authentication system provides flexible credential management with support for both driver-level and session-level authentication.

Capabilities

Authentication Functions

Factory functions for creating authentication tokens for different authentication schemes.

def basic_auth(user: str, password: str, realm: str = None) -> Auth:
    """
    Create basic authentication token for username/password authentication.
    
    Parameters:
    - user: Username for authentication
    - password: Password for authentication  
    - realm: Authentication realm (optional)
    
    Returns:
    Auth token configured for basic authentication
    """

def kerberos_auth(base64_encoded_ticket: str) -> Auth:
    """
    Create Kerberos authentication token.
    
    Parameters:
    - base64_encoded_ticket: Base64 encoded Kerberos ticket
    
    Returns:
    Auth token configured for Kerberos authentication
    """

def bearer_auth(base64_encoded_token: str) -> Auth:
    """
    Create bearer token authentication for OAuth/JWT scenarios.
    
    Parameters:
    - base64_encoded_token: Base64 encoded bearer token
    
    Returns:
    Auth token configured for bearer token authentication
    """

def custom_auth(
    principal: str,
    credentials: str, 
    realm: str,
    scheme: str,
    **parameters
) -> Auth:
    """
    Create custom authentication token for non-standard schemes.
    
    Parameters:
    - principal: User being authenticated
    - credentials: Authentication credentials
    - realm: Authentication realm/provider
    - scheme: Authentication scheme name
    - **parameters: Additional authentication parameters
    
    Returns:
    Auth token configured with custom authentication details
    """

Example authentication usage:

from neo4j import GraphDatabase, basic_auth, bearer_auth, custom_auth

# Basic username/password authentication (most common)
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("neo4j", "password")
)

# Basic auth with realm
driver = GraphDatabase.driver(
    "bolt://localhost:7687", 
    auth=basic_auth("user", "password", realm="production")
)

# Bearer token authentication (OAuth/JWT)
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=bearer_auth("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...")
)

# Custom authentication scheme  
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=custom_auth(
        principal="user@domain.com",
        credentials="secret_key", 
        realm="ldap_server",
        scheme="LDAP",
        server="ldap.company.com",
        port=389
    )
)

Auth Class

Container class holding authentication details and credentials for database connections.

class Auth:
    def __init__(
        self,
        scheme: str,
        principal: str, 
        credentials: str,
        realm: str = None,
        **parameters
    ):
        """
        Create authentication token.
        
        Parameters:
        - scheme: Authentication scheme identifier
        - principal: User being authenticated
        - credentials: Authentication credentials (password, token, etc.)
        - realm: Authentication provider/realm
        - **parameters: Additional scheme-specific parameters
        """
    
    @property
    def scheme(self) -> str | None:
        """
        Authentication scheme identifier.
        
        Returns:
        String identifying the authentication method (basic, kerberos, etc.)
        """
    
    @property
    def principal(self) -> str | None:
        """
        User being authenticated.
        
        Returns:
        Username or principal identifier
        """
    
    @property
    def credentials(self) -> str | None:
        """
        Authentication credentials.
        
        Returns:
        Password, token, or other credential data
        """
    
    @property
    def realm(self) -> str | None:
        """
        Authentication provider or realm.
        
        Returns:
        Realm identifier for the authentication provider
        """
    
    @property
    def parameters(self) -> dict:
        """
        Additional authentication parameters.
        
        Returns:
        Dictionary of scheme-specific parameters
        """

AuthToken is an alias for the Auth class:

# AuthToken is identical to Auth
AuthToken = Auth

Authentication Examples

Basic Authentication

The most common authentication method using username and password:

from neo4j import GraphDatabase, basic_auth

# Simple basic authentication
auth = basic_auth("neo4j", "password")
driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

# With explicit realm
auth = basic_auth("username", "password", realm="my_realm")
driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

# Manual Auth construction (equivalent to basic_auth)
from neo4j import Auth
auth = Auth("basic", "neo4j", "password")
driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

Session-Level Authentication

Authentication can be overridden at the session level for per-session credentials:

# Driver with default authentication
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("default_user", "default_password")
)

# Session with different authentication
with driver.session(auth=basic_auth("admin_user", "admin_password")) as session:
    # This session uses admin credentials
    result = session.run("SHOW USERS")
    
# Back to default credentials for subsequent sessions
with driver.session() as session:
    # Uses default_user credentials
    result = session.run("MATCH (n) RETURN count(n)")

Bearer Token Authentication

For OAuth, JWT, or other token-based authentication:

from neo4j import GraphDatabase, bearer_auth

# Using bearer token (JWT example)
jwt_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"

auth = bearer_auth(jwt_token)
driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

# Token refresh example
class TokenRefreshAuth:
    def __init__(self, initial_token):
        self._token = initial_token
    
    def get_current_auth(self):
        # Refresh token logic here
        return bearer_auth(self._token)

token_auth = TokenRefreshAuth("initial_token")
driver = GraphDatabase.driver("bolt://localhost:7687", auth=token_auth.get_current_auth())

Kerberos Authentication

For environments using Kerberos authentication:

from neo4j import GraphDatabase, kerberos_auth
import base64

# Obtain Kerberos ticket (implementation specific)
kerberos_ticket = obtain_kerberos_ticket()  # Your Kerberos implementation
encoded_ticket = base64.b64encode(kerberos_ticket).decode('utf-8')

auth = kerberos_auth(encoded_ticket)
driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

Custom Authentication Schemes

For proprietary or specialized authentication systems:

from neo4j import GraphDatabase, custom_auth

# LDAP authentication example
auth = custom_auth(
    principal="user@company.com",
    credentials="ldap_password",
    realm="ldap.company.com", 
    scheme="LDAP",
    # Additional LDAP-specific parameters
    server="ldap.company.com",
    port=389,
    base_dn="ou=users,dc=company,dc=com"
)

driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

# API key authentication example
auth = custom_auth(
    principal="api_user",
    credentials="api_key_12345",
    realm="api_service",
    scheme="APIKEY",
    # Additional parameters
    service_name="neo4j_service",
    key_version="v2"
)

driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth)

No Authentication

For development or unsecured environments:

from neo4j import GraphDatabase

# No authentication (auth=None is default)
driver = GraphDatabase.driver("bolt://localhost:7687")

# Explicit no auth
driver = GraphDatabase.driver("bolt://localhost:7687", auth=None)

Authentication Verification

Drivers can verify authentication credentials before executing queries:

from neo4j import GraphDatabase, basic_auth

driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))

try:
    # Verify authentication works
    is_valid = driver.verify_authentication()
    print(f"Authentication valid: {is_valid}")
    
    # Verify different credentials
    alt_auth = basic_auth("different_user", "different_password")
    is_valid = driver.verify_authentication(auth=alt_auth)
    print(f"Alternative auth valid: {is_valid}")
    
except Exception as e:
    print(f"Authentication verification failed: {e}")

Security Considerations

Credential Storage

import os
from neo4j import GraphDatabase, basic_auth

# Use environment variables for credentials
username = os.getenv("NEO4J_USERNAME", "neo4j")
password = os.getenv("NEO4J_PASSWORD")

if not password:
    raise ValueError("NEO4J_PASSWORD environment variable must be set")

driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth(username, password)
)

Connection Encryption

Authentication should be used with encrypted connections:

from neo4j import GraphDatabase, basic_auth

# Secure connection with authentication
driver = GraphDatabase.driver(
    "bolt+s://production.neo4j.com:7687",  # Encrypted connection
    auth=basic_auth("username", "password"),
    encrypted=True,
    trust=neo4j.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES
)

Authentication Rotation

class RotatingAuth:
    def __init__(self):
        self._current_auth = None
        self._refresh_time = 0
    
    def get_auth(self):
        import time
        now = time.time()
        
        # Refresh every hour
        if now - self._refresh_time > 3600:
            self._current_auth = self._refresh_credentials()
            self._refresh_time = now
        
        return self._current_auth
    
    def _refresh_credentials(self):
        # Implement credential refresh logic
        new_token = fetch_new_token()
        return bearer_auth(new_token)

auth_manager = RotatingAuth()
driver = GraphDatabase.driver("bolt://localhost:7687", auth=auth_manager.get_auth())

Install with Tessl CLI