chore(security): drop passlib for direct bcrypt + reject CLI-flag check targets

- Replace passlib.CryptContext with bcrypt 4.2.1 directly (passlib is
  unmaintained and only emitted warnings when paired with bcrypt 4+).
  hash_password / verify_password now call bcrypt.checkpw / .hashpw, and
  verify_password is hardened against empty inputs without raising.
  scripts/hash_password.py + tests/conftest.py updated to match.
- status_checker.check_node() now rejects targets starting with '-' before
  any subprocess invocation, defending ping/tcp paths against arg-injection
  if an admin sets a check_target like '-O'.

Tests added:
- test_auth: expired JWT, malformed JWT, wrong-secret JWT, empty password
  vs empty server hash, verify_password edge cases.
- test_scan: _background_scan failure path marks ScanRun failed, leaves
  non-running terminal status alone, success path invokes run_scan.
- test_status_checker: ping/tcp invocations are bypassed when target or
  ip starts with '-'.

Backend coverage 89% → 90%.
This commit is contained in:
Pouzor
2026-05-14 12:11:40 +02:00
parent 5f87c64dcf
commit aa17edf1d0
8 changed files with 199 additions and 17 deletions
+6 -6
View File
@@ -1,22 +1,22 @@
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
import bcrypt
from jose import JWTError, jwt from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain: str, hashed: str) -> bool: def verify_password(plain: str, hashed: str) -> bool:
if not plain or not hashed:
return False
try: try:
return bool(pwd_context.verify(plain, hashed)) return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8"))
except ValueError: except (ValueError, TypeError):
return False return False
def hash_password(password: str) -> str: def hash_password(password: str) -> str:
return str(pwd_context.hash(password)) return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
def create_access_token(subject: str) -> str: def create_access_token(subject: str) -> str:
+5
View File
@@ -24,6 +24,11 @@ async def check_node(check_method: str, target: str | None, ip: str | None) -> d
host = target or raw_ip host = target or raw_ip
if not host: if not host:
return {"status": "unknown", "response_time_ms": None} return {"status": "unknown", "response_time_ms": None}
# Reject hostnames that look like CLI flags — defends ping/tcp invocations
# against arg-injection if a malicious admin sets target like "-O".
if host.startswith("-"):
logger.warning("Rejecting check target that starts with '-': %r", host)
return {"status": "unknown", "response_time_ms": None}
start = time.monotonic() start = time.monotonic()
try: try:
+1 -2
View File
@@ -7,8 +7,7 @@ alembic==1.13.3
pydantic==2.9.2 pydantic==2.9.2
pydantic-settings==2.5.2 pydantic-settings==2.5.2
python-jose[cryptography]==3.5.0 python-jose[cryptography]==3.5.0
passlib[bcrypt]==1.7.4 bcrypt==4.2.1
bcrypt==4.0.1
python-multipart==0.0.27 python-multipart==0.0.27
apscheduler==3.10.4 apscheduler==3.10.4
python-nmap==0.7.1 python-nmap==0.7.1
+3 -5
View File
@@ -1,13 +1,11 @@
"""Generate a bcrypt password hash for config.yml.""" """Generate a bcrypt password hash for the AUTH_PASSWORD_HASH env var."""
import sys import sys
from passlib.context import CryptContext import bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
if len(sys.argv) < 2: if len(sys.argv) < 2:
print("Usage: python scripts/hash_password.py <password>") print("Usage: python scripts/hash_password.py <password>")
sys.exit(1) sys.exit(1)
password = sys.argv[1] password = sys.argv[1]
print(pwd_context.hash(password)) print(bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8"))
+2 -4
View File
@@ -5,23 +5,21 @@ os.environ.setdefault("SECRET_KEY", "test-only-secret-key-not-for-production")
import pytest import pytest
from httpx import ASGITransport, AsyncClient from httpx import ASGITransport, AsyncClient
from passlib.context import CryptContext
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.core.security import hash_password
from app.db.database import Base, get_db from app.db.database import Base, get_db
from app.main import app from app.main import app
TEST_DB_URL = "sqlite+aiosqlite:///:memory:" TEST_DB_URL = "sqlite+aiosqlite:///:memory:"
_pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
@pytest.fixture(autouse=True, scope="session") @pytest.fixture(autouse=True, scope="session")
def test_credentials(): def test_credentials():
"""Configure test auth credentials directly on settings.""" """Configure test auth credentials directly on settings."""
from app.core.config import settings from app.core.config import settings
settings.auth_username = "admin" settings.auth_username = "admin"
settings.auth_password_hash = _pwd_ctx.hash("admin") settings.auth_password_hash = hash_password("admin")
@pytest.fixture @pytest.fixture
+68
View File
@@ -68,3 +68,71 @@ async def test_login_with_malformed_hash_returns_401_not_500(client: AsyncClient
assert res.status_code == 401 assert res.status_code == 401
finally: finally:
settings.auth_password_hash = original settings.auth_password_hash = original
# --- JWT-level cases ---
async def test_expired_token_rejected(client: AsyncClient):
"""A JWT whose `exp` is in the past must be refused."""
from datetime import datetime, timedelta, timezone
from jose import jwt
from app.core.config import settings
payload = {
"sub": "admin",
"exp": datetime.now(timezone.utc) - timedelta(minutes=1),
}
token = jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
res = await client.get("/api/v1/nodes", headers={"Authorization": f"Bearer {token}"})
assert res.status_code == 401
async def test_malformed_token_rejected(client: AsyncClient):
res = await client.get("/api/v1/nodes", headers={"Authorization": "Bearer not-a-jwt"})
assert res.status_code == 401
async def test_token_signed_with_wrong_secret_rejected(client: AsyncClient):
"""A token signed with a different key must not be accepted."""
from datetime import datetime, timedelta, timezone
from jose import jwt
from app.core.config import settings
payload = {
"sub": "admin",
"exp": datetime.now(timezone.utc) + timedelta(minutes=5),
}
forged = jwt.encode(payload, "different-secret", algorithm=settings.algorithm)
res = await client.get("/api/v1/nodes", headers={"Authorization": f"Bearer {forged}"})
assert res.status_code == 401
async def test_missing_authorization_header_rejected(client: AsyncClient):
res = await client.get("/api/v1/nodes")
assert res.status_code == 401
async def test_empty_password_does_not_pass_when_hash_empty(client: AsyncClient):
"""No credentials configured server-side must not authenticate an empty password."""
from app.core.config import settings
original_hash = settings.auth_password_hash
settings.auth_password_hash = ""
try:
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": ""})
assert res.status_code == 401
finally:
settings.auth_password_hash = original_hash
# --- Password helper ---
def test_verify_password_handles_empty_inputs():
"""verify_password must be safe against empty plain / empty hash without raising."""
from app.core.security import hash_password, verify_password
h = hash_password("hunter2")
assert verify_password("hunter2", h) is True
assert verify_password("", h) is False
assert verify_password("hunter2", "") is False
assert verify_password("", "") is False
+89
View File
@@ -37,6 +37,95 @@ async def pending_device(db_session):
return device return device
# --- _background_scan error handling ---
@pytest.fixture
async def mem_db():
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.db.database import Base
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
factory = async_sessionmaker(engine, expire_on_commit=False)
yield factory
await engine.dispose()
@pytest.mark.asyncio
async def test_background_scan_marks_run_failed_on_exception(mem_db):
"""If run_scan() raises, the ScanRun must transition running → failed and the
session rollback path must execute without a follow-on exception."""
from app.api.routes.scan import _background_scan
async with mem_db() as session:
run = ScanRun(status="running", ranges=["10.0.0.0/24"])
session.add(run)
await session.commit()
run_id = run.id
with (
patch("app.api.routes.scan.AsyncSessionLocal", mem_db),
patch(
"app.api.routes.scan.run_scan",
new_callable=AsyncMock,
side_effect=RuntimeError("boom"),
),
):
await _background_scan(run_id, ["10.0.0.0/24"])
async with mem_db() as session:
refreshed = await session.get(ScanRun, run_id)
assert refreshed is not None
assert refreshed.status == "failed"
@pytest.mark.asyncio
async def test_background_scan_leaves_non_running_status_alone(mem_db):
"""If the run was already stopped/cancelled before run_scan failed, _background_scan
must NOT overwrite that terminal status with 'failed'."""
from app.api.routes.scan import _background_scan
async with mem_db() as session:
run = ScanRun(status="cancelled", ranges=["10.0.0.0/24"])
session.add(run)
await session.commit()
run_id = run.id
with (
patch("app.api.routes.scan.AsyncSessionLocal", mem_db),
patch(
"app.api.routes.scan.run_scan",
new_callable=AsyncMock,
side_effect=RuntimeError("boom"),
),
):
await _background_scan(run_id, ["10.0.0.0/24"])
async with mem_db() as session:
refreshed = await session.get(ScanRun, run_id)
assert refreshed is not None
assert refreshed.status == "cancelled"
@pytest.mark.asyncio
async def test_background_scan_success_path_invokes_run_scan(mem_db):
from app.api.routes.scan import _background_scan
async with mem_db() as session:
run = ScanRun(status="running", ranges=["10.0.0.0/24"])
session.add(run)
await session.commit()
run_id = run.id
with (
patch("app.api.routes.scan.AsyncSessionLocal", mem_db),
patch("app.api.routes.scan.run_scan", new_callable=AsyncMock) as mock_run_scan,
):
await _background_scan(run_id, ["10.0.0.0/24"])
mock_run_scan.assert_awaited_once()
# --- Trigger scan --- # --- Trigger scan ---
@pytest.mark.asyncio @pytest.mark.asyncio
+25
View File
@@ -216,6 +216,31 @@ async def test_ping_uses_windows_args_on_win32():
assert "-c" not in captured["args"] assert "-c" not in captured["args"]
# --- check_node target validation ---
@pytest.mark.asyncio
async def test_check_node_rejects_flag_like_target():
"""A target starting with '-' must never reach subprocess invocation."""
from app.services.status_checker import check_node
with patch("asyncio.create_subprocess_exec") as mock_exec:
result = await check_node("ping", "-O", None)
mock_exec.assert_not_called()
assert result["status"] == "unknown"
@pytest.mark.asyncio
async def test_check_node_rejects_flag_like_ip():
from app.services.status_checker import check_node
with patch("asyncio.create_subprocess_exec") as mock_exec:
result = await check_node("ping", None, "-O")
mock_exec.assert_not_called()
assert result["status"] == "unknown"
# --- _tcp_connect --- # --- _tcp_connect ---
@pytest.mark.asyncio @pytest.mark.asyncio