Security & Authentication

# Enabling access control — the correct sequence

from pymongo import MongoClient

# ── STEP 1: Connect WITHOUT authentication (only possible before enabling) ─
client_no_auth = MongoClient("mongodb://localhost:27017/")
admin_db = client_no_auth["admin"]

# ── STEP 2: Create the first admin user BEFORE enabling access control ──────
# Must be done in the admin database
# roles: userAdminAnyDatabase — can manage users across all databases
#        readWriteAnyDatabase  — can read and write to any database
#        clusterAdmin          — can manage the replica set / cluster

admin_db.command("createUser", "dataplexa_admin",
    pwd="StrongP@ssword123!",
    roles=[
        {"role": "userAdminAnyDatabase", "db": "admin"},
        {"role": "readWriteAnyDatabase", "db": "admin"},
        {"role": "clusterAdmin",         "db": "admin"},
    ]
)
print("Admin user created: dataplexa_admin")

# ── STEP 3: Enable access control in mongod.conf ─────────────────────────
# Edit /etc/mongod.conf and add:
mongod_conf_security = """
# /etc/mongod.conf — security section
security:
  authorization: enabled
"""
print("\nmongod.conf security section:")
print(mongod_conf_security)

# ── STEP 4: Restart MongoDB ───────────────────────────────────────────────
print("sudo systemctl restart mongod")
print()

# ── STEP 5: Connect WITH authentication ──────────────────────────────────
client_auth = MongoClient(
    "mongodb://dataplexa_admin:StrongP@ssword123!@localhost:27017/",
    authSource="admin"    # the database where the user was created
)
print("Authenticated connection successful:")
print(f"  server info: {client_auth.admin.command('serverStatus')['host']}")

# Verify authentication is enforced
print("\nAuthentication enforcement check:")
try:
    client_unauth = MongoClient(
        "mongodb://localhost:27017/",
        serverSelectionTimeoutMS=3000
    )
    client_unauth.admin.command("listDatabases")
    print("  ✗ Unauthenticated connection succeeded — access control NOT enabled")
except Exception as e:
    print(f"  ✓ Unauthenticated connection rejected: {type(e).__name__}")

# SCRAM authentication — creating purpose-built application users

from pymongo import MongoClient
import os

# Connect as admin to create application users
client = MongoClient(
    "mongodb://dataplexa_admin:StrongP@ssword123!@localhost:27017/",
    authSource="admin"
)
admin_db = client["admin"]

# ── Application user — read/write on dataplexa only ───────────────────────
# This is what your application server uses
admin_db.command("createUser", "app_user",
    pwd=os.environ.get("APP_DB_PASSWORD", "AppP@ss999!"),
    roles=[
        {"role": "readWrite", "db": "dataplexa"}   # only dataplexa, nothing else
    ]
)
print("Created: app_user  (readWrite on dataplexa only)")

# ── Read-only user — for reporting and analytics ──────────────────────────
admin_db.command("createUser", "report_user",
    pwd=os.environ.get("REPORT_DB_PASSWORD", "ReportP@ss777!"),
    roles=[
        {"role": "read", "db": "dataplexa"}   # read-only
    ]
)
print("Created: report_user  (read-only on dataplexa)")

# ── Monitoring user — for ops tools like MongoDB Compass or Grafana ───────
admin_db.command("createUser", "monitor_user",
    pwd=os.environ.get("MONITOR_DB_PASSWORD", "MonitorP@ss555!"),
    roles=[
        {"role": "clusterMonitor", "db": "admin"},
        {"role": "read",           "db": "local"}
    ]
)
print("Created: monitor_user  (cluster monitoring only)")

# ── List all users ────────────────────────────────────────────────────────
print("\nAll database users:")
users = admin_db.command("usersInfo", 1)  # 1 = all users
for user in users.get("users", []):
    roles = [f"{r['role']}@{r['db']}" for r in user.get("roles", [])]
    print(f"  {user['user']:15}  roles: {', '.join(roles)}")

# ── Connect as app_user and verify privileges ─────────────────────────────
app_client = MongoClient(
    "mongodb://app_user:AppP@ss999!@localhost:27017/dataplexa",
    authSource="dataplexa"
)
app_db = app_client["dataplexa"]

# This should work — app_user has readWrite on dataplexa
app_db.products.find_one({}, {"name": 1})
print("\napp_user read from dataplexa.products:  ✓ allowed")

# This should fail — app_user has no access to admin database
try:
    app_client["admin"].command("listDatabases")
    print("app_user listed admin databases:  ✗ should have been blocked")
except Exception:
    print("app_user access to admin database:  ✓ correctly blocked")

# TLS configuration — enabling encrypted connections

from pymongo import MongoClient
import ssl

# ── mongod.conf — enable TLS ──────────────────────────────────────────────
tls_conf = """
# /etc/mongod.conf — TLS section
net:
  tls:
    mode: requireTLS              # reject all non-TLS connections
    certificateKeyFile: /etc/ssl/mongodb/server.pem    # server cert + key
    CAFile: /etc/ssl/mongodb/ca.pem                    # CA certificate
    allowConnectionsWithoutCertificates: true          # client cert optional
"""
print("mongod.conf TLS configuration:")
print(tls_conf)

# ── Connect with TLS — certificate verification ───────────────────────────
print("TLS connection scenarios:\n")

# Scenario 1: TLS with CA verification (production standard)
print("1. Production — TLS with CA certificate verification:")
print("""
   client = MongoClient(
       "mongodb://app_user:pass@server:27017/dataplexa",
       tls=True,
       tlsCAFile="/path/to/ca.pem",      # verify server cert against this CA
       authSource="dataplexa"
   )
""")

# Scenario 2: TLS with mutual authentication (x.509)
print("2. Mutual TLS — both server and client present certificates:")
print("""
   client = MongoClient(
       "mongodb://server:27017/dataplexa",
       tls=True,
       tlsCAFile="/path/to/ca.pem",
       tlsCertificateKeyFile="/path/to/client.pem",  # client presents its cert
       authMechanism="MONGODB-X509"                  # authenticate via cert CN
   )
""")

# Scenario 3: TLS in development — allow self-signed cert (NOT for production)
print("3. Development only — allow self-signed / unverified cert:")
print("""
   client = MongoClient(
       "mongodb://localhost:27017/",
       tls=True,
       tlsAllowInvalidCertificates=True   # ← NEVER use in production
   )
""")

# TLS mode options
print("TLS mode options (mongod.conf net.tls.mode):\n")
modes = [
    ("disabled",        "No TLS — plaintext only (default, never use in production)"),
    ("allowTLS",        "Accept both TLS and non-TLS connections"),
    ("preferTLS",       "Prefer TLS but accept non-TLS — for migration periods"),
    ("requireTLS",      "Reject all non-TLS connections (production standard)"),
]
for mode, desc in modes:
    prefix = "✓" if mode == "requireTLS" else "○"
    print(f"  {prefix} {mode:16}  {desc}")

# Network hardening — bindIP, firewall rules, and auditing open connections

from pymongo import MongoClient

# ── mongod.conf — network binding ────────────────────────────────────────
net_conf = """
# /etc/mongod.conf — network section (hardened)
net:
  port: 27017
  bindIp: 127.0.0.1,10.0.1.50   # localhost + specific private IP only
  # bindIpAll: false             # NEVER set bindIpAll: true in production
  tls:
    mode: requireTLS
    certificateKeyFile: /etc/ssl/mongodb/server.pem
    CAFile: /etc/ssl/mongodb/ca.pem
"""
print("mongod.conf network hardening:")
print(net_conf)

# ── What bindIp controls ──────────────────────────────────────────────────
print("bindIp explained:\n")
bindings = [
    ("127.0.0.1",           "Localhost only — no remote connections possible"),
    ("127.0.0.1,10.0.1.50", "Localhost + one private IP — app server only"),
    ("0.0.0.0",             "All interfaces — exposed to public network (dangerous)"),
]
for binding, desc in bindings:
    flag = "✓" if "10.0.1" in binding or binding == "127.0.0.1" else "✗"
    print(f"  {flag} bindIp: {binding:25}  {desc}")

# ── Firewall rules (iptables example) ────────────────────────────────────
print("\nLinux firewall rules (iptables):")
fw_rules = [
    "# Allow MongoDB only from app server IP",
    "iptables -A INPUT -p tcp --dport 27017 -s 10.0.1.50 -j ACCEPT",
    "# Allow localhost",
    "iptables -A INPUT -p tcp --dport 27017 -s 127.0.0.1 -j ACCEPT",
    "# Block all other MongoDB connections",
    "iptables -A INPUT -p tcp --dport 27017 -j DROP",
]
for rule in fw_rules:
    print(f"  {rule}")

# ── Check current connections ─────────────────────────────────────────────
client = MongoClient(
    "mongodb://dataplexa_admin:StrongP@ssword123!@localhost:27017/",
    authSource="admin"
)
print("\nCurrent active connections:")
server_status = client.admin.command("serverStatus")
connections = server_status.get("connections", {})
print(f"  current:   {connections.get('current', 0)}")
print(f"  available: {connections.get('available', 0)}")
print(f"  total:     {connections.get('totalCreated', 0)}")

# List active operations — find any suspicious long-running ops
print("\nActive operations (currentOp):")
current_ops = client.admin.command("currentOp", {"active": True})
for op in current_ops.get("inprog", [])[:5]:
    print(f"  opid: {op.get('opid'):6}  "
          f"type: {op.get('type','?'):10}  "
          f"client: {op.get('client','?'):20}  "
          f"secs: {op.get('secs_running', 0)}")

# Security audit — automated checks from PyMongo

from pymongo import MongoClient

client = MongoClient(
    "mongodb://dataplexa_admin:StrongP@ssword123!@localhost:27017/",
    authSource="admin"
)
admin_db = client["admin"]

print("MongoDB Security Audit\n" + "═" * 40)

issues   = []
warnings = []

# ── Check 1: Authentication enabled ──────────────────────────────────────
cmd_line = client.admin.command("getCmdLineOpts")
auth_enabled = cmd_line.get("parsed", {}).get(
    "security", {}
).get("authorization", "") == "enabled"

status = "✓" if auth_enabled else "✗"
print(f"\n{status} Authentication: {'ENABLED' if auth_enabled else 'DISABLED'}")
if not auth_enabled:
    issues.append("Authentication is disabled — enable security.authorization in mongod.conf")

# ── Check 2: TLS mode ─────────────────────────────────────────────────────
tls_mode = cmd_line.get("parsed", {}).get("net", {}).get("tls", {}).get("mode", "disabled")
tls_ok   = tls_mode == "requireTLS"
status   = "✓" if tls_ok else "✗"
print(f"{status} TLS mode: {tls_mode}")
if not tls_ok:
    issues.append(f"TLS mode is '{tls_mode}' — set to requireTLS in production")

# ── Check 3: Users with excessive privileges ──────────────────────────────
print("\nUser privilege audit:")
all_users = admin_db.command("usersInfo", 1).get("users", [])
high_priv_roles = {"root", "dbOwner", "__system", "userAdminAnyDatabase"}
for user in all_users:
    user_roles = {r["role"] for r in user.get("roles", [])}
    excess = user_roles & high_priv_roles
    if excess and user["user"] != "dataplexa_admin":
        warnings.append(f"User '{user['user']}' has high-privilege roles: {excess}")
        print(f"  ⚠ {user['user']:15}  high-privilege roles: {excess}")
    else:
        print(f"  ✓ {user['user']:15}  roles look appropriate")

# ── Check 4: Default port in use ─────────────────────────────────────────
port = cmd_line.get("parsed", {}).get("net", {}).get("port", 27017)
port_ok = port != 27017
status  = "○" if not port_ok else "✓"
print(f"\n{status} Port: {port} {'(non-default — good)' if port_ok else '(default 27017 — consider changing)'}")

# ── Check 5: No users with empty passwords ────────────────────────────────
print("\nEmpty password check: ✓ not checkable via driver — verify in mongod logs")

# ── Summary ───────────────────────────────────────────────────────────────
print(f"\n{'═'*40}")
print(f"Issues:   {len(issues)}")
for i in issues:
    print(f"  ✗ {i}")
print(f"Warnings: {len(warnings)}")
for w in warnings:
    print(f"  ⚠ {w}")
if not issues and not warnings:
    print("  ✓ All security checks passed")