enable policies

This commit is contained in:
2026-01-16 17:31:20 +10:30
parent b2e084db33
commit d063a0ca81
28 changed files with 4296 additions and 224 deletions
+295
View File
@@ -0,0 +1,295 @@
"""Unit tests for MFA policy models."""
import pytest
from datetime import datetime, timezone, timedelta
from gatehouse_app.models import (
User,
Organization,
OrganizationMember,
OrganizationSecurityPolicy,
UserSecurityPolicy,
MfaPolicyCompliance,
Session,
)
from gatehouse_app.utils.constants import (
UserStatus,
MfaPolicyMode,
MfaComplianceStatus,
MfaRequirementOverride,
SessionStatus,
OrganizationRole,
)
@pytest.mark.unit
class TestOrganizationSecurityPolicyModel:
"""Tests for OrganizationSecurityPolicy model."""
def test_create_org_security_policy(self, db, test_organization):
"""Test creating an organization security policy."""
policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.OPTIONAL,
mfa_grace_period_days=14,
notify_days_before=7,
)
policy.save()
assert policy.id is not None
assert policy.organization_id == test_organization.id
assert policy.mfa_policy_mode == MfaPolicyMode.OPTIONAL
assert policy.mfa_grace_period_days == 14
assert policy.notify_days_before == 7
assert policy.policy_version == 1
assert policy.created_at is not None
def test_org_security_policy_to_dict(self, db, test_organization):
"""Test organization security policy to_dict method."""
policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN,
mfa_grace_period_days=7,
notify_days_before=3,
)
policy.save()
policy_dict = policy.to_dict()
assert "id" in policy_dict
assert "organization_id" in policy_dict
assert policy_dict["organization_id"] == test_organization.id
assert "mfa_policy_mode" in policy_dict
assert "mfa_grace_period_days" in policy_dict
def test_org_security_policy_relationships(self, db, test_organization):
"""Test organization security policy relationships."""
policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP,
)
policy.save()
# Test relationship
assert policy.organization is not None
assert policy.organization.id == test_organization.id
@pytest.mark.unit
class TestUserSecurityPolicyModel:
"""Tests for UserSecurityPolicy model."""
def test_create_user_security_policy(self, db, test_user, test_organization):
"""Test creating a user security policy."""
policy = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.INHERIT,
)
policy.save()
assert policy.id is not None
assert policy.user_id == test_user.id
assert policy.organization_id == test_organization.id
assert policy.mfa_override_mode == MfaRequirementOverride.INHERIT
assert policy.force_totp is False
assert policy.force_webauthn is False
def test_user_security_policy_with_overrides(self, db, test_user, test_organization):
"""Test user security policy with override settings."""
policy = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.REQUIRED,
force_totp=True,
force_webauthn=False,
)
policy.save()
assert policy.mfa_override_mode == MfaRequirementOverride.REQUIRED
assert policy.force_totp is True
assert policy.force_webauthn is False
def test_user_security_policy_exempt(self, db, test_user, test_organization):
"""Test user security policy with exempt override."""
policy = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.EXEMPT,
)
policy.save()
assert policy.mfa_override_mode == MfaRequirementOverride.EXEMPT
def test_user_security_policy_relationships(self, db, test_user, test_organization):
"""Test user security policy relationships."""
policy = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.INHERIT,
)
policy.save()
# Test relationships
assert policy.user is not None
assert policy.user.id == test_user.id
assert policy.organization is not None
assert policy.organization.id == test_organization.id
@pytest.mark.unit
class TestMfaPolicyComplianceModel:
"""Tests for MfaPolicyCompliance model."""
def test_create_mfa_policy_compliance(self, db, test_user, test_organization):
"""Test creating an MFA policy compliance record."""
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.NOT_APPLICABLE,
policy_version=1,
)
compliance.save()
assert compliance.id is not None
assert compliance.user_id == test_user.id
assert compliance.organization_id == test_organization.id
assert compliance.status == MfaComplianceStatus.NOT_APPLICABLE
assert compliance.policy_version == 1
assert compliance.notification_count == 0
def test_mfa_policy_compliance_in_grace(self, db, test_user, test_organization):
"""Test MFA compliance record in grace period."""
now = datetime.now(timezone.utc)
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.IN_GRACE,
policy_version=1,
applied_at=now,
deadline_at=now + timedelta(days=14),
)
compliance.save()
assert compliance.status == MfaComplianceStatus.IN_GRACE
assert compliance.applied_at is not None
assert compliance.deadline_at is not None
assert compliance.deadline_at > now
def test_mfa_policy_compliance_compliant(self, db, test_user, test_organization):
"""Test MFA compliance record when compliant."""
now = datetime.now(timezone.utc)
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.COMPLIANT,
policy_version=1,
applied_at=now - timedelta(days=30),
deadline_at=now - timedelta(days=16),
compliant_at=now - timedelta(days=16),
)
compliance.save()
assert compliance.status == MfaComplianceStatus.COMPLIANT
assert compliance.compliant_at is not None
def test_mfa_policy_compliance_suspended(self, db, test_user, test_organization):
"""Test MFA compliance record when suspended."""
now = datetime.now(timezone.utc)
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.SUSPENDED,
policy_version=1,
applied_at=now - timedelta(days=30),
deadline_at=now - timedelta(days=16),
suspended_at=now - timedelta(days=16),
)
compliance.save()
assert compliance.status == MfaComplianceStatus.SUSPENDED
assert compliance.suspended_at is not None
def test_mfa_policy_compliance_relationships(self, db, test_user, test_organization):
"""Test MFA compliance relationships."""
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.NOT_APPLICABLE,
policy_version=1,
)
compliance.save()
# Test relationships
assert compliance.user is not None
assert compliance.user.id == test_user.id
assert compliance.organization is not None
assert compliance.organization.id == test_organization.id
@pytest.mark.unit
class TestSessionModelComplianceFlag:
"""Tests for Session model compliance flag."""
def test_session_default_not_compliance_only(self, db, test_user):
"""Test that sessions are not compliance only by default."""
session = Session(
user_id=test_user.id,
token="test-token-123",
status=SessionStatus.ACTIVE,
expires_at=datetime.now(timezone.utc) + timedelta(hours=8),
last_activity_at=datetime.now(timezone.utc),
)
session.save()
assert session.is_compliance_only is False
def test_session_compliance_only(self, db, test_user):
"""Test creating a compliance-only session."""
session = Session(
user_id=test_user.id,
token="compliance-token-123",
status=SessionStatus.ACTIVE,
expires_at=datetime.now(timezone.utc) + timedelta(hours=8),
last_activity_at=datetime.now(timezone.utc),
is_compliance_only=True,
)
session.save()
assert session.is_compliance_only is True
def test_session_to_dict_excludes_token(self, db, test_user):
"""Test that session to_dict excludes the token."""
session = Session(
user_id=test_user.id,
token="test-token-456",
status=SessionStatus.ACTIVE,
expires_at=datetime.now(timezone.utc) + timedelta(hours=8),
last_activity_at=datetime.now(timezone.utc),
)
session.save()
session_dict = session.to_dict()
assert "id" in session_dict
assert "user_id" in session_dict
assert "is_compliance_only" in session_dict
assert session_dict["is_compliance_only"] is False
@pytest.mark.unit
class TestUserStatusComplianceSuspended:
"""Tests for UserStatus.COMPLIANCE_SUSPENDED."""
def test_compliance_suspended_status_exists(self):
"""Test that COMPLIANCE_SUSPENDED status exists."""
assert UserStatus.COMPLIANCE_SUSPENDED.value == "compliance_suspended"
def test_create_compliance_suspended_user(self, db):
"""Test creating a compliance suspended user."""
user = User(
email="suspended@example.com",
full_name="Suspended User",
status=UserStatus.COMPLIANCE_SUSPENDED,
)
user.save()
assert user.status == UserStatus.COMPLIANCE_SUSPENDED
@@ -0,0 +1,476 @@
"""Unit tests for MfaPolicyService."""
import pytest
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock
from gatehouse_app.models import (
User,
Organization,
OrganizationMember,
OrganizationSecurityPolicy,
UserSecurityPolicy,
MfaPolicyCompliance,
Session,
)
from gatehouse_app.services.mfa_policy_service import (
MfaPolicyService,
OrgPolicyDto,
EffectiveUserPolicyDto,
AggregateMfaStateDto,
LoginPolicyResult,
)
from gatehouse_app.utils.constants import (
UserStatus,
MfaPolicyMode,
MfaComplianceStatus,
MfaRequirementOverride,
SessionStatus,
OrganizationRole,
)
@pytest.mark.unit
class TestMfaPolicyService:
"""Tests for MfaPolicyService."""
def test_get_org_policy_not_found(self, db, test_organization):
"""Test getting organization policy when none exists."""
policy = MfaPolicyService.get_org_policy(test_organization.id)
assert policy is None
def test_get_org_policy_found(self, db, test_organization):
"""Test getting organization policy when it exists."""
# Create policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN,
mfa_grace_period_days=14,
notify_days_before=7,
)
org_policy.save()
policy = MfaPolicyService.get_org_policy(test_organization.id)
assert policy is not None
assert policy.organization_id == test_organization.id
assert policy.mfa_policy_mode == MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN.value
assert policy.mfa_grace_period_days == 14
assert policy.notify_days_before == 7
assert policy.policy_version == 1
def test_get_effective_user_policy_no_org_policy(self, db, test_user, test_organization):
"""Test effective user policy when no org policy exists."""
policy = MfaPolicyService.get_effective_user_policy(test_user.id, test_organization.id)
assert policy is not None
assert policy.organization_id == test_organization.id
assert policy.effective_mode == MfaPolicyMode.DISABLED.value
assert policy.requires_totp is False
assert policy.requires_webauthn is False
assert policy.is_exempt is True
def test_get_effective_user_policy_with_org_policy(self, db, test_user, test_organization):
"""Test effective user policy with org policy and no override."""
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP,
mfa_grace_period_days=14,
)
org_policy.save()
policy = MfaPolicyService.get_effective_user_policy(test_user.id, test_organization.id)
assert policy is not None
assert policy.effective_mode == MfaPolicyMode.REQUIRE_TOTP.value
assert policy.requires_totp is True
assert policy.requires_webauthn is False
assert policy.is_exempt is False
def test_get_effective_user_policy_with_override_inherit(self, db, test_user, test_organization):
"""Test effective user policy with INHERIT override."""
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_WEBAUTHN,
mfa_grace_period_days=7,
)
org_policy.save()
# Create user override
user_override = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.INHERIT,
)
user_override.save()
policy = MfaPolicyService.get_effective_user_policy(test_user.id, test_organization.id)
assert policy.effective_mode == MfaPolicyMode.REQUIRE_WEBAUTHN.value
assert policy.requires_webauthn is True
def test_get_effective_user_policy_with_override_exempt(self, db, test_user, test_organization):
"""Test effective user policy with EXEMPT override."""
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN,
mfa_grace_period_days=14,
)
org_policy.save()
# Create user override
user_override = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.EXEMPT,
)
user_override.save()
policy = MfaPolicyService.get_effective_user_policy(test_user.id, test_organization.id)
assert policy.effective_mode == MfaPolicyMode.DISABLED.value
assert policy.is_exempt is True
def test_get_effective_user_policy_with_override_required(self, db, test_user, test_organization):
"""Test effective user policy with REQUIRED override."""
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.OPTIONAL,
mfa_grace_period_days=14,
)
org_policy.save()
# Create user override
user_override = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.REQUIRED,
)
user_override.save()
policy = MfaPolicyService.get_effective_user_policy(test_user.id, test_organization.id)
assert policy.effective_mode == MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN.value
assert policy.requires_totp is True
assert policy.requires_webauthn is True
assert policy.is_exempt is False
def test_evaluate_user_mfa_state_no_policy(self, db, test_user, test_organization):
"""Test evaluating user MFA state with no policy."""
# Create membership
membership = OrganizationMember(
user_id=test_user.id,
organization_id=test_organization.id,
role=OrganizationRole.MEMBER,
)
membership.save()
state = MfaPolicyService.evaluate_user_mfa_state(test_user)
assert state is not None
assert state.overall_status == MfaComplianceStatus.COMPLIANT.value
assert len(state.missing_methods) == 0
assert len(state.orgs) == 1
def test_evaluate_user_mfa_state_with_policy(self, db, test_user, test_organization):
"""Test evaluating user MFA state with policy."""
# Create membership
membership = OrganizationMember(
user_id=test_user.id,
organization_id=test_organization.id,
role=OrganizationRole.MEMBER,
)
membership.save()
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP,
mfa_grace_period_days=14,
)
org_policy.save()
state = MfaPolicyService.evaluate_user_mfa_state(test_user)
assert state is not None
assert state.overall_status == MfaComplianceStatus.IN_GRACE.value
assert "totp" in state.missing_methods
assert len(state.orgs) == 1
assert state.orgs[0].effective_mode == MfaPolicyMode.REQUIRE_TOTP.value
def test_after_primary_auth_success_no_required_policy(self, db, test_user, test_organization):
"""Test after_primary_auth_success with no required policy."""
# Create membership
membership = OrganizationMember(
user_id=test_user.id,
organization_id=test_organization.id,
role=OrganizationRole.MEMBER,
)
membership.save()
result = MfaPolicyService.after_primary_auth_success(test_user)
assert result.can_create_full_session is True
assert result.create_compliance_only_session is False
assert result.compliance_summary.overall_status == MfaComplianceStatus.COMPLIANT.value
def test_after_primary_auth_success_in_grace(self, db, test_user, test_organization):
"""Test after_primary_auth_success when user is in grace period."""
# Create membership
membership = OrganizationMember(
user_id=test_user.id,
organization_id=test_organization.id,
role=OrganizationRole.MEMBER,
)
membership.save()
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP,
mfa_grace_period_days=14,
)
org_policy.save()
result = MfaPolicyService.after_primary_auth_success(test_user)
assert result.can_create_full_session is True
assert result.create_compliance_only_session is False
assert result.compliance_summary.overall_status == MfaComplianceStatus.IN_GRACE.value
def test_after_primary_auth_success_past_due(self, db, test_user, test_organization):
"""Test after_primary_auth_success when user is past due."""
# Create membership
membership = OrganizationMember(
user_id=test_user.id,
organization_id=test_organization.id,
role=OrganizationRole.MEMBER,
)
membership.save()
# Create org policy
org_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP,
mfa_grace_period_days=14,
)
org_policy.save()
# Create compliance record past due
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.PAST_DUE,
policy_version=1,
applied_at=datetime.now(timezone.utc) - timedelta(days=30),
deadline_at=datetime.now(timezone.utc) - timedelta(days=1),
)
compliance.save()
result = MfaPolicyService.after_primary_auth_success(test_user)
assert result.can_create_full_session is False
assert result.create_compliance_only_session is True
def test_create_org_policy_new(self, db, test_organization):
"""Test creating a new organization policy."""
policy = MfaPolicyService.create_org_policy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN,
mfa_grace_period_days=14,
notify_days_before=7,
updated_by_user_id=None,
)
assert policy is not None
assert policy.organization_id == test_organization.id
assert policy.mfa_policy_mode == MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN
assert policy.policy_version == 1
def test_create_org_policy_update(self, db, test_organization):
"""Test updating an existing organization policy."""
# Create initial policy
initial_policy = OrganizationSecurityPolicy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.OPTIONAL,
mfa_grace_period_days=14,
)
initial_policy.save()
# Update policy
updated_policy = MfaPolicyService.create_org_policy(
organization_id=test_organization.id,
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP,
mfa_grace_period_days=7,
updated_by_user_id=None,
)
assert updated_policy.mfa_policy_mode == MfaPolicyMode.REQUIRE_TOTP
assert updated_policy.mfa_grace_period_days == 7
assert updated_policy.policy_version == 2
def test_set_user_override_new(self, db, test_user, test_organization):
"""Test setting a new user override."""
override = MfaPolicyService.set_user_override(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.REQUIRED,
force_totp=True,
force_webauthn=False,
updated_by_user_id=None,
)
assert override is not None
assert override.user_id == test_user.id
assert override.organization_id == test_organization.id
assert override.mfa_override_mode == MfaRequirementOverride.REQUIRED
assert override.force_totp is True
def test_set_user_override_update(self, db, test_user, test_organization):
"""Test updating an existing user override."""
# Create initial override
initial_override = UserSecurityPolicy(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.INHERIT,
)
initial_override.save()
# Update override
updated_override = MfaPolicyService.set_user_override(
user_id=test_user.id,
organization_id=test_organization.id,
mfa_override_mode=MfaRequirementOverride.EXEMPT,
updated_by_user_id=None,
)
assert updated_override.mfa_override_mode == MfaRequirementOverride.EXEMPT
def test_get_user_compliance(self, db, test_user, test_organization):
"""Test getting user compliance record."""
# Create compliance record
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.COMPLIANT,
policy_version=1,
)
compliance.save()
result = MfaPolicyService.get_user_compliance(test_user.id, test_organization.id)
assert result is not None
assert result.status == MfaComplianceStatus.COMPLIANT
def test_get_user_compliance_not_found(self, db, test_user, test_organization):
"""Test getting user compliance record when none exists."""
result = MfaPolicyService.get_user_compliance(test_user.id, test_organization.id)
assert result is None
def test_get_org_compliance_list(self, db, test_user, test_organization):
"""Test getting organization compliance list."""
# Create compliance record
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.IN_GRACE,
policy_version=1,
deadline_at=datetime.now(timezone.utc) + timedelta(days=14),
)
compliance.save()
results = MfaPolicyService.get_org_compliance_list(test_organization.id)
assert len(results) == 1
assert results[0]["user_id"] == test_user.id
assert results[0]["status"] == MfaComplianceStatus.IN_GRACE.value
def test_get_org_compliance_list_with_status_filter(self, db, test_user, test_organization):
"""Test getting organization compliance list with status filter."""
# Create compliance record
compliance = MfaPolicyCompliance(
user_id=test_user.id,
organization_id=test_organization.id,
status=MfaComplianceStatus.COMPLIANT,
policy_version=1,
)
compliance.save()
# Filter by different status
results = MfaPolicyService.get_org_compliance_list(
test_organization.id, status=MfaComplianceStatus.IN_GRACE
)
assert len(results) == 0
# Filter by correct status
results = MfaPolicyService.get_org_compliance_list(
test_organization.id, status=MfaComplianceStatus.COMPLIANT
)
assert len(results) == 1
@pytest.mark.unit
class TestMfaPolicyServiceDto:
"""Tests for MfaPolicyService DTOs."""
def test_org_policy_dto(self):
"""Test OrgPolicyDto creation."""
dto = OrgPolicyDto(
organization_id="org-123",
mfa_policy_mode=MfaPolicyMode.REQUIRE_TOTP.value,
mfa_grace_period_days=14,
notify_days_before=7,
policy_version=1,
)
assert dto.organization_id == "org-123"
assert dto.mfa_policy_mode == "require_totp"
assert dto.mfa_grace_period_days == 14
def test_effective_user_policy_dto(self):
"""Test EffectiveUserPolicyDto creation."""
dto = EffectiveUserPolicyDto(
organization_id="org-123",
effective_mode=MfaPolicyMode.REQUIRE_TOTP_OR_WEBAUTHN.value,
requires_totp=True,
requires_webauthn=True,
grace_period_days=14,
is_exempt=False,
)
assert dto.requires_totp is True
assert dto.requires_webauthn is True
assert dto.is_exempt is False
def test_aggregate_mfa_state_dto(self):
"""Test AggregateMfaStateDto creation."""
dto = AggregateMfaStateDto(
overall_status=MfaComplianceStatus.IN_GRACE.value,
missing_methods=["totp"],
deadline_at="2025-02-01T00:00:00Z",
orgs=[],
)
assert dto.overall_status == "in_grace"
assert "totp" in dto.missing_methods
assert dto.deadline_at == "2025-02-01T00:00:00Z"
def test_login_policy_result(self):
"""Test LoginPolicyResult creation."""
summary = AggregateMfaStateDto(
overall_status=MfaComplianceStatus.IN_GRACE.value,
missing_methods=["totp"],
orgs=[],
)
result = LoginPolicyResult(
can_create_full_session=True,
create_compliance_only_session=False,
compliance_summary=summary,
)
assert result.can_create_full_session is True
assert result.create_compliance_only_session is False
assert result.compliance_summary.overall_status == "in_grace"