🧪 Lab 10: Security¶

Topics: Users, roles, custom roles, SCRAM-SHA-256

---

In [ ]:

from pymongo import MongoClient, ReadPreference
from pymongo.read_concern import ReadConcern
from pymongo.write_concern import WriteConcern
from pymongo.errors import ConnectionFailure
from bson import ObjectId
from datetime import datetime, timedelta, timezone
import pprint

USE_ATLAS = False
ATLAS_URI  = "mongodb+srv://<username>:<password>@<cluster>.mongodb.net/?retryWrites=true&w=majority"
DOCKER_URI = "mongodb://127.0.0.1:27017/?directConnection=true"

uri = ATLAS_URI if USE_ATLAS else DOCKER_URI
client = MongoClient(uri, serverSelectionTimeoutMS=5000)
try:
    client.admin.command("ping")
    print("✅ Connected to", "Atlas" if USE_ATLAS else "Docker MongoDB")
except ConnectionFailure as e:
    print("❌ Connection failed:", e)

db = client["mongo_labs"]
pp = pprint.PrettyPrinter(indent=2)

from pymongo import MongoClient, ReadPreference from pymongo.read_concern import ReadConcern from pymongo.write_concern import WriteConcern from pymongo.errors import ConnectionFailure from bson import ObjectId from datetime import datetime, timedelta, timezone import pprint USE_ATLAS = False ATLAS_URI = "mongodb+srv://<username>:<password>@<cluster>.mongodb.net/?retryWrites=true&w=majority" DOCKER_URI = "mongodb://127.0.0.1:27017/?directConnection=true" uri = ATLAS_URI if USE_ATLAS else DOCKER_URI client = MongoClient(uri, serverSelectionTimeoutMS=5000) try: client.admin.command("ping") print("✅ Connected to", "Atlas" if USE_ATLAS else "Docker MongoDB") except ConnectionFailure as e: print("❌ Connection failed:", e) db = client["mongo_labs"] pp = pprint.PrettyPrinter(indent=2)

Create Users¶

In [ ]:

admin_db = client["admin"]
try:
    admin_db.command("createUser","admin",pwd="AdminPass123!",roles=[{"role":"root","db":"admin"}])
    print("✅ Created admin")
except Exception as e:
    print(f"ℹ️  {e}")

admin_db = client["admin"] try: admin_db.command("createUser","admin",pwd="AdminPass123!",roles=[{"role":"root","db":"admin"}]) print("✅ Created admin") except Exception as e: print(f"ℹ️ {e}")

In [ ]:

try:
    admin_db.command("createUser","app_user",pwd="AppPass456!",roles=[{"role":"readWrite","db":"mongo_labs"}],mechanisms=["SCRAM-SHA-256"])
    print("✅ Created app_user")
except Exception as e:
    print(f"ℹ️  {e}")

try: admin_db.command("createUser","app_user",pwd="AppPass456!",roles=[{"role":"readWrite","db":"mongo_labs"}],mechanisms=["SCRAM-SHA-256"]) print("✅ Created app_user") except Exception as e: print(f"ℹ️ {e}")

In [ ]:

try:
    admin_db.command("createUser","reporter",pwd="Report789!",roles=[{"role":"read","db":"mongo_labs"}])
    print("✅ Created reporter")
except Exception as e:
    print(f"ℹ️  {e}")

try: admin_db.command("createUser","reporter",pwd="Report789!",roles=[{"role":"read","db":"mongo_labs"}]) print("✅ Created reporter") except Exception as e: print(f"ℹ️ {e}")

Custom Role¶

In [ ]:

mongo_labs_db = client["mongo_labs"]
try:
    mongo_labs_db.command("createRole","ordersReader",privileges=[{"resource":{"db":"mongo_labs","collection":"orders"},"actions":["find"]}],roles=[])
    print("✅ Created ordersReader role")
except Exception as e:
    print(f"ℹ️  {e}")

mongo_labs_db = client["mongo_labs"] try: mongo_labs_db.command("createRole","ordersReader",privileges=[{"resource":{"db":"mongo_labs","collection":"orders"},"actions":["find"]}],roles=[]) print("✅ Created ordersReader role") except Exception as e: print(f"ℹ️ {e}")

In [ ]:

try:
    admin_db.command("createUser","orders_svc",pwd="OrdersPass!",roles=[{"role":"ordersReader","db":"mongo_labs"}])
    print("✅ Created orders_svc with ordersReader")
except Exception as e:
    print(f"ℹ️  {e}")

try: admin_db.command("createUser","orders_svc",pwd="OrdersPass!",roles=[{"role":"ordersReader","db":"mongo_labs"}]) print("✅ Created orders_svc with ordersReader") except Exception as e: print(f"ℹ️ {e}")

List Users¶

In [ ]:

users = admin_db.command("usersInfo")
for u in users.get("users",[]):
    roles = [f"{r['role']}@{r['db']}" for r in u.get("roles",[])]
    print(f"  {u['user']:20s}: {roles}")

users = admin_db.command("usersInfo") for u in users.get("users",[]): roles = [f"{r['role']}@{r['db']}" for r in u.get("roles",[])] print(f" {u['user']:20s}: {roles}")

Grant/Revoke Roles¶

In [ ]:

admin_db.command("grantRolesToUser","reporter",roles=[{"role":"readWrite","db":"mongo_labs"}])
print("✅ Granted readWrite to reporter")
admin_db.command("revokeRolesFromUser","reporter",roles=[{"role":"readWrite","db":"mongo_labs"}])
print("✅ Revoked readWrite from reporter")

admin_db.command("grantRolesToUser","reporter",roles=[{"role":"readWrite","db":"mongo_labs"}]) print("✅ Granted readWrite to reporter") admin_db.command("revokeRolesFromUser","reporter",roles=[{"role":"readWrite","db":"mongo_labs"}]) print("✅ Revoked readWrite from reporter")