darkstrata-credential-check 1.0.0

Check if credentials have been exposed in data breaches using k-anonymity

darkstrata-credential-check

Check if credentials have been exposed in data breaches using k-anonymity.

Features

• Privacy-first: Uses k-anonymity to check credentials without exposing them
• Type-safe: Full type hints with comprehensive dataclass definitions
• Async-first: Built on httpx for efficient async HTTP operations
• Automatic caching: Intelligent caching aligned with server time windows
• Batch processing: Efficiently check multiple credentials with automatic prefix grouping
• Retry logic: Built-in exponential backoff for transient failures
• Comprehensive errors: Detailed error types for easy error handling

Requirements

• Python 3.10 or higher

Installation

pip install darkstrata-credential-check

poetry add darkstrata-credential-check

uv add darkstrata-credential-check

Quick Start

import asyncio
from darkstrata_credential_check import DarkStrataCredentialCheck

async def main():
    # Create a client
    async with DarkStrataCredentialCheck(api_key='your-api-key') as client:
        # Check a single credential
        result = await client.check('[e-mail]', 'password123')

        if result.found:
            print('WARNING: This credential was found in a data breach!')
        else:
            print('Credential not found in known breaches.')

asyncio.run(main())

How It Works

This SDK uses k-anonymity to check credentials without exposing them:

1. Your credential is hashed locally: SHA256(email:password)
2. Only the first 5 characters (prefix) of the hash are sent to the API
3. The API returns all hashes matching that prefix (1-in-1,000,000 anonymity)
4. The SDK checks if your full hash is in the returned set

Your actual credentials never leave your system.

┌─────────────────────┐         ┌─────────────────────┐
│     Your System     │         │   DarkStrata API    │
│                     │         │                     │
│  email:password     │         │                     │
│        ↓            │         │                     │
│  SHA256 hash        │         │                     │
│        ↓            │         │                     │
│  Extract prefix ────┼────────→│  Lookup by prefix   │
│  (5 chars only)     │         │        ↓            │
│                     │←────────┼─ Return all matches │
│  Check membership   │         │                     │
│        ↓            │         │                     │
│  found: true/false  │         │                     │
└─────────────────────┘         └─────────────────────┘

API Reference

DarkStrataCredentialCheck

The main client class.

Constructor

DarkStrataCredentialCheck(
    api_key: str,
    *,
    base_url: str = 'https://api.darkstrata.io/v1/',
    timeout: float = 30.0,
    retries: int = 3,
    enable_caching: bool = True,
    cache_ttl: int = 3600,
)

Parameters:

Parameter	Type	Default	Description
api_key	str	required	Your DarkStrata API key
base_url	str	'https://api.darkstrata.io/v1/'	API base URL
timeout	float	30.0	Request timeout in seconds
retries	int	3	Number of retry attempts
enable_caching	bool	True	Enable response caching
cache_ttl	int	3600	Cache TTL in seconds (1 hour)

Methods

check(email, password, options?)

Check a single credential.

result = await client.check('[e-mail]', 'password123')

Returns: CheckResult

check_hash(hash, options?)

Check a pre-computed SHA-256 hash.

hash_value = '5BAA61E4C9B93F3F0682250B6CF8331B7EE68FD8...'
result = await client.check_hash(hash_value)

Returns: CheckResult

check_batch(credentials, options?)

Check multiple credentials efficiently.

from darkstrata_credential_check import Credential

results = await client.check_batch([
    Credential(email='[e-mail]', password='pass1'),
    Credential(email='[e-mail]', password='pass2'),
])

Returns: list[CheckResult]

clear_cache()

Clear the internal response cache.

client.clear_cache()

get_cache_size()

Get the number of cached entries.

size = client.get_cache_size()

CheckResult

The result of a credential check.

@dataclass
class CheckResult:
    found: bool                    # true if credential was in a breach
    credential: CredentialInfo     # info about the credential checked
    metadata: CheckMetadata        # additional metadata

CheckMetadata

Metadata returned with check results.

@dataclass
class CheckMetadata:
    prefix: str                    # The 5-char prefix used
    total_results: int             # Total hashes returned by API
    hmac_source: Literal['server', 'client']  # Source of HMAC key
    time_window: int | None        # Server time window (server HMAC only)
    filter_since: int | None       # Epoch day filter (if since was used)
    cached_result: bool            # Whether result was from cache
    checked_at: datetime           # When the check was performed

CheckOptions

Optional parameters for check requests.

@dataclass
class CheckOptions:
    client_hmac: str | None = None   # Your own HMAC key (64+ hex chars)
    since: int | datetime | None = None  # Filter by breach date

Error Handling

The SDK provides specific error types for different failure scenarios:

from darkstrata_credential_check import (
    DarkStrataCredentialCheck,
    AuthenticationError,
    ValidationError,
    ApiError,
    TimeoutError,
    NetworkError,
    RateLimitError,
    is_darkstrata_error,
)

async with DarkStrataCredentialCheck(api_key='your-key') as client:
    try:
        result = await client.check('[e-mail]', 'password')
    except AuthenticationError:
        print('Invalid API key')
    except ValidationError as error:
        print(f'Invalid input: {error.field}')
    except RateLimitError as error:
        if error.retry_after:
            print(f'Rate limited. Retry after {error.retry_after} seconds')
    except TimeoutError as error:
        print(f'Request timed out after {error.timeout_seconds}s')
    except NetworkError as error:
        print(f'Network error: {error}')
    except ApiError as error:
        print(f'API error ({error.status_code}): {error}')
    except Exception as error:
        if is_darkstrata_error(error):
            print(f'DarkStrata error [{error.code}]: {error}')
        else:
            raise

Error Types

Error	Code	Description
AuthenticationError	AUTHENTICATION_ERROR	Invalid or missing API key
ValidationError	VALIDATION_ERROR	Invalid input parameters
ApiError	API_ERROR	API request failed
TimeoutError	TIMEOUT_ERROR	Request timed out
NetworkError	NETWORK_ERROR	Network connectivity issue
RateLimitError	RATE_LIMIT_ERROR	Rate limit exceeded

Advanced Usage

Pre-computed Hashes

If you're storing hashed credentials, you can check them directly:

from darkstrata_credential_check import hash_credential, DarkStrataCredentialCheck

# Compute hash once
hash_value = hash_credential('[e-mail]', 'password123')
# Store hash securely...

# Later, check the hash
async with DarkStrataCredentialCheck(api_key='your-key') as client:
    result = await client.check_hash(hash_value)

Batch Processing

For checking multiple credentials, use check_batch for efficiency:

from darkstrata_credential_check import Credential

credentials = [
    Credential(email='[e-mail]', password='pass1'),
    Credential(email='[e-mail]', password='pass2'),
    Credential(email='[e-mail]', password='pass3'),
]

results = await client.check_batch(credentials)

compromised = [r for r in results if r.found]
print(f'{len(compromised)} credentials were compromised')

Batch processing automatically groups credentials by prefix to minimise API calls.

Client-Provided HMAC Key

By default, the server generates a time-rotating HMAC key. For deterministic results across requests, provide your own key:

import secrets
from darkstrata_credential_check import CheckOptions

# Generate a secure key once and store it securely
client_hmac = secrets.token_hex(32)  # 64 hex chars = 256 bits

result = await client.check('[e-mail]', 'password',
    CheckOptions(client_hmac=client_hmac))

# Results are now deterministic (not time-windowed)
print(result.metadata.hmac_source)  # 'client'

When to use client HMAC:

• You need consistent results across multiple requests
• You're comparing results from different time periods
• You want to avoid server-side key rotation

Date Filtering

Filter results to only include breaches from a specific date onwards:

from datetime import datetime
from darkstrata_credential_check import CheckOptions

# Only check breaches from 2024 onwards
result = await client.check('[e-mail]', 'password',
    CheckOptions(since=datetime(2024, 1, 1)))

# Or use epoch day (days since 1 January 1970)
result = await client.check('[e-mail]', 'password',
    CheckOptions(since=19724))  # 2024-01-01

# Check the filter applied
print(result.metadata.filter_since)  # 19724

Combined Options

You can combine multiple options:

result = await client.check('[e-mail]', 'password',
    CheckOptions(
        client_hmac='your-256-bit-hex-key...',
        since=datetime(2024, 1, 1),
    ))

Disabling Cache

For real-time checks where you need fresh results:

client = DarkStrataCredentialCheck(
    api_key='your-key',
    enable_caching=False,
)

Custom Timeout and Retries

client = DarkStrataCredentialCheck(
    api_key='your-key',
    timeout=60.0,  # 60 seconds
    retries=5,     # 5 retry attempts
)

Security Considerations

What is sent to the API?

• Only the first 5 characters of the SHA-256 hash
• Your API key for authentication

What is NOT sent?

• Your email address
• Your password
• The full hash of your credentials

Best Practices

1. Never log credentials - The SDK never logs credentials, and you shouldn't either
2. Use HTTPS - The SDK enforces HTTPS for all API calls
3. Secure your API key - Store your API key securely (environment variables, secrets manager)
4. Handle errors gracefully - Don't expose internal errors to end users

Type Hints

This package includes comprehensive type hints and exports all types:

from darkstrata_credential_check import (
    ClientOptions,
    Credential,
    CheckOptions,
    CheckResult,
    CheckMetadata,
    CredentialInfo,
)

Context Manager

The client supports async context manager for proper resource cleanup:

async with DarkStrataCredentialCheck(api_key='your-key') as client:
    result = await client.check('[e-mail]', 'password')
# HTTP client is automatically closed

Alternatively, close manually:

client = DarkStrataCredentialCheck(api_key='your-key')
try:
    result = await client.check('[e-mail]', 'password')
finally:
    await client.close()

Contributing

See the contributing guide.

Licence

Apache 2.0 (c) DarkStrata