diff --git a/gatehouse_app/api/v1/zerotier.py b/gatehouse_app/api/v1/zerotier.py index a2d5535..ebd8776 100644 --- a/gatehouse_app/api/v1/zerotier.py +++ b/gatehouse_app/api/v1/zerotier.py @@ -301,7 +301,7 @@ def get_network_members(org_id, network_id): memberships = portal_network_service.get_network_members(network_id) return api_response( - data={"memberships": [m.to_dict() for m in memberships], "count": len(memberships)}, + data={"memberships": memberships, "count": len(memberships)}, message="Network members retrieved successfully", ) diff --git a/gatehouse_app/services/portal_network_service.py b/gatehouse_app/services/portal_network_service.py index 7a710e7..db283a3 100644 --- a/gatehouse_app/services/portal_network_service.py +++ b/gatehouse_app/services/portal_network_service.py @@ -308,11 +308,33 @@ def get_network_members(network_id: str) -> list: from gatehouse_app.models import NetworkAccessRequest from gatehouse_app.utils.constants import ApprovalState - return NetworkAccessRequest.query.filter( - NetworkAccessRequest.portal_network_id == network_id, - NetworkAccessRequest.status == ApprovalState.APPROVED, - NetworkAccessRequest.deleted_at.is_(None), - ).order_by(NetworkAccessRequest.created_at.desc()).all() + members = ( + NetworkAccessRequest.query + .options( + db.joinedload(NetworkAccessRequest.user), + db.joinedload(NetworkAccessRequest.device), + ) + .filter( + NetworkAccessRequest.portal_network_id == network_id, + NetworkAccessRequest.status == ApprovalState.APPROVED, + NetworkAccessRequest.deleted_at.is_(None), + ) + .order_by(NetworkAccessRequest.created_at.desc()) + .all() + ) + + result = [] + for m in members: + d = m.to_dict() + user = m.user + device = m.device + d["user_email"] = user.email if user else None + d["user_name"] = user.full_name if user else None + d["device_name"] = device.display_name if device else None + d["device_node_id"] = device.node_id if device else None + result.append(d) + + return result def get_network_pending_requests(network_id: str) -> list: diff --git a/tests/integration/test_zerotier.py b/tests/integration/test_zerotier.py index 541c908..0d780b3 100644 --- a/tests/integration/test_zerotier.py +++ b/tests/integration/test_zerotier.py @@ -605,6 +605,457 @@ class TestAdminForceDeleteMembership: ) assert exc_info.value.status_code == 404 +class TestZeroTierNetworkMembers: + """Test GET /organizations//networks//members. + + Ensures the endpoint returns human-friendly names (user_email, user_name, + device_name, device_node_id) instead of bare UUIDs. + """ + + def _setup_network_and_member( + self, integration_app, org_id, user_id, device_kwargs=None + ): + """Helper: create a PortalNetwork and an approved NetworkAccessRequest. + + Returns (network_id, request_id). + """ + from gatehouse_app.models.zerotier.device import Device + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode + + kwargs = { + "node_id": "a1b2c3d4e5", + "device_nickname": "My Laptop", + "hostname": "my-laptop", + } + if device_kwargs: + kwargs.update(device_kwargs) + + with integration_app.app_context(): + device = Device( + user_id=user_id, + organization_id=org_id, + **{k: v for k, v in kwargs.items() if hasattr(Device, k)}, + ) + _db.session.add(device) + _db.session.flush() + device_id = device.id + + network = PortalNetwork( + organization_id=org_id, + name="Test Net", + owner_user_id=user_id, + zerotier_network_id="bbbbbbbbbb", + request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.flush() + network_id = network.id + + request = NetworkAccessRequest( + organization_id=org_id, + user_id=user_id, + device_id=device_id, + portal_network_id=network_id, + status=ApprovalState.APPROVED, + active=True, + justification="Need access for work", + ) + _db.session.add(request) + _db.session.commit() + return network_id, device_id, request.id + + def test_empty_when_no_members( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-01: Network with no approved members returns empty list. + + Creates a network with no NetworkAccessRequests at all. + """ + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import NetworkRequestMode + + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + with integration_app.app_context(): + network = PortalNetwork( + organization_id=org["id"], + name="Empty Net", + owner_user_id=user["id"], + zerotier_network_id="cccccccccc", + request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.commit() + net_id = network.id + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data.get("memberships") == [] + assert data.get("count") == 0 + + def test_returns_enriched_fields( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-02: Approved members include user_email, user_name, device_name, device_node_id. + + Verifies the full enrichment pipeline with a user that has both + email and full_name and a device with all identifiers. + """ + user = create_test_user( + password="Pass1234!", + email="enriched_alice@test.com", + full_name="Alice Smith", + ) + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + net_id, device_id, request_id = self._setup_network_and_member( + integration_app, org["id"], user["id"], + {"device_nickname": "Alice Laptop", "hostname": "alice-laptop", "node_id": "aaaaaaaaaa"}, + ) + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + + assert data["count"] == 1 + member = data["memberships"][0] + assert member["user_email"] == "enriched_alice@test.com" + assert member["user_name"] == "Alice Smith" + assert member["device_name"] == "Alice Laptop" + assert member["device_node_id"] == "aaaaaaaaaa" + assert member["user_id"] == user["id"] + assert member["device_id"] == device_id + assert "id" in member + assert member["status"] == "approved" + assert member["justification"] == "Need access for work" + + def test_device_name_fallback_hostname( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-03: device_name falls back to hostname when device_nickname is None.""" + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + net_id, _, _ = self._setup_network_and_member( + integration_app, org["id"], user["id"], + {"device_nickname": None, "hostname": "headless-server", "node_id": "bbbbbbbbbb"}, + ) + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 1 + assert data["memberships"][0]["device_name"] == "headless-server" + + def test_device_name_fallback_node_id( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-04: device_name falls back to node_id when nickname and hostname are both None.""" + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + net_id, _, _ = self._setup_network_and_member( + integration_app, org["id"], user["id"], + {"device_nickname": None, "hostname": None, "node_id": "cccccccccc"}, + ) + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 1 + assert data["memberships"][0]["device_name"] == "cccccccccc" + + def test_user_name_is_none_when_missing( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-05: user_name is None when the user record has no full_name.""" + user = create_test_user(password="Pass1234!", full_name=None) + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + net_id, _, _ = self._setup_network_and_member( + integration_app, org["id"], user["id"], + ) + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 1 + assert data["memberships"][0]["user_name"] is None + assert data["memberships"][0]["user_email"] == user["email"] + + def test_multiple_members_returned( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-06: Multiple approved members for the same network all appear. + + Two users, each with their own device, both approved on the same network. + """ + from gatehouse_app.models.zerotier.device import Device + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode + + admin = create_test_user(password="Pass1234!") + user_a = create_test_user(password="Pass1234!", email="multi_alice@test.com", full_name="Alice") + user_b = create_test_user(password="Pass1234!", email="multi_bob@test.com", full_name="Bob") + org = create_test_org() + create_test_membership(admin["id"], org["id"]) + create_test_membership(user_a["id"], org["id"]) + create_test_membership(user_b["id"], org["id"]) + + with integration_app.app_context(): + network = PortalNetwork( + organization_id=org["id"], + name="Shared Net", + owner_user_id=admin["id"], + zerotier_network_id="dddddddddd", + request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.flush() + net_id = network.id + + dev_a = Device(user_id=user_a["id"], organization_id=org["id"], + node_id="device_a_01", device_nickname="Alice Phone") + _db.session.add(dev_a) + _db.session.flush() + dev_b = Device(user_id=user_b["id"], organization_id=org["id"], + node_id="device_b_01", device_nickname="Bob Tablet") + _db.session.add(dev_b) + _db.session.flush() + + req_a = NetworkAccessRequest( + organization_id=org["id"], user_id=user_a["id"], device_id=dev_a.id, + portal_network_id=net_id, status=ApprovalState.APPROVED, active=True, + ) + req_b = NetworkAccessRequest( + organization_id=org["id"], user_id=user_b["id"], device_id=dev_b.id, + portal_network_id=net_id, status=ApprovalState.APPROVED, active=True, + ) + _db.session.add_all([req_a, req_b]) + _db.session.commit() + + integration_client.auth.login(email=admin["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 2 + + emails = {m["user_email"] for m in data["memberships"]} + names = {m["user_name"] for m in data["memberships"]} + device_names = {m["device_name"] for m in data["memberships"]} + assert emails == {"multi_alice@test.com", "multi_bob@test.com"} + assert names == {"Alice", "Bob"} + assert device_names == {"Alice Phone", "Bob Tablet"} + + def test_pending_memberships_excluded( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-07: Pending (non-approved) memberships are NOT returned. + + Only APPROVED status memberships should appear. + """ + from gatehouse_app.models.zerotier.device import Device + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode + + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + with integration_app.app_context(): + network = PortalNetwork( + organization_id=org["id"], name="Test", owner_user_id=user["id"], + zerotier_network_id="eeeeeeeeee", request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.flush() + net_id = network.id + + device = Device( + user_id=user["id"], organization_id=org["id"], + node_id="ffffffffff", device_nickname="Pending Device", + ) + _db.session.add(device) + _db.session.flush() + + req = NetworkAccessRequest( + organization_id=org["id"], user_id=user["id"], device_id=device.id, + portal_network_id=net_id, status=ApprovalState.PENDING, active=False, + ) + _db.session.add(req) + _db.session.commit() + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 0 + assert data["memberships"] == [] + + def test_soft_deleted_memberships_excluded( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-08: Soft-deleted (deleted_at set) memberships are not returned.""" + from datetime import datetime, timezone + from gatehouse_app.models.zerotier.device import Device + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode + + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + with integration_app.app_context(): + network = PortalNetwork( + organization_id=org["id"], name="Test", owner_user_id=user["id"], + zerotier_network_id="gggggggggg", request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.flush() + net_id = network.id + + device = Device( + user_id=user["id"], organization_id=org["id"], + node_id="hhhhhhhhhh", device_nickname="Deleted Device", + ) + _db.session.add(device) + _db.session.flush() + + req = NetworkAccessRequest( + organization_id=org["id"], user_id=user["id"], device_id=device.id, + portal_network_id=net_id, status=ApprovalState.APPROVED, active=True, + deleted_at=datetime.now(timezone.utc), + ) + _db.session.add(req) + _db.session.commit() + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 0 + + def test_network_not_found_negative( + self, integration_client, create_test_user, create_test_org, create_test_membership + ): + """ZT-MEM-09: 404 when network_id does not exist in the organization.""" + import uuid + + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + integration_client.auth.login(email=user["email"], password="Pass1234!") + with pytest.raises(ApiError) as exc: + integration_client.get( + f"/organizations/{org['id']}/networks/{uuid.uuid4()}/members" + ) + assert exc.value.status_code == 404 + + def test_user_not_in_org_negative( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-10: 403 when the requesting user is not a member of the org.""" + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import NetworkRequestMode + + owner = create_test_user(password="Owner1234!") + intruder = create_test_user(password="Intruder1234!") + org = create_test_org() + + # Only the owner is a member; intruder is NOT in this org + create_test_membership(owner["id"], org["id"], OrganizationRole.ADMIN) + + with integration_app.app_context(): + network = PortalNetwork( + organization_id=org["id"], name="Secret Net", owner_user_id=owner["id"], + zerotier_network_id="iiiiiiiiii", request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.commit() + net_id = network.id + + integration_client.auth.login(email=intruder["email"], password="Intruder1234!") + with pytest.raises(ApiError) as exc: + integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + assert exc.value.status_code == 403 + + def test_auth_required_negative(self, integration_client): + """ZT-MEM-11: 401 when no auth token is provided.""" + import uuid + + with pytest.raises(ApiError) as exc: + integration_client.get( + f"/organizations/{uuid.uuid4()}/networks/{uuid.uuid4()}/members" + ) + assert exc.value.status_code == 401 + + def test_no_active_session( + self, integration_client, create_test_user, create_test_org, create_test_membership, integration_app + ): + """ZT-MEM-12: Inactive (active=False) approved memberships are still returned. + + The endpoint returns ALL approved memberships regardless of the + `active` flag — both active=True and active=False appear. + """ + from gatehouse_app.models.zerotier.device import Device + from gatehouse_app.models.zerotier.portal_network import PortalNetwork + from gatehouse_app.models.zerotier.network_access_request import NetworkAccessRequest + from gatehouse_app.extensions import db as _db + from gatehouse_app.utils.constants import ApprovalState, NetworkRequestMode + + user = create_test_user(password="Pass1234!") + org = create_test_org() + create_test_membership(user["id"], org["id"]) + + with integration_app.app_context(): + network = PortalNetwork( + organization_id=org["id"], name="Test", owner_user_id=user["id"], + zerotier_network_id="jjjjjjjjjj", request_mode=NetworkRequestMode.OPEN, + ) + _db.session.add(network) + _db.session.flush() + net_id = network.id + + device = Device( + user_id=user["id"], organization_id=org["id"], + node_id="kkkkkkkkkk", device_nickname="Inactive Dev", + ) + _db.session.add(device) + _db.session.flush() + + req = NetworkAccessRequest( + organization_id=org["id"], user_id=user["id"], device_id=device.id, + portal_network_id=net_id, status=ApprovalState.APPROVED, active=False, + ) + _db.session.add(req) + _db.session.commit() + + integration_client.auth.login(email=user["email"], password="Pass1234!") + result = integration_client.get(f"/organizations/{org['id']}/networks/{net_id}/members") + data = result.get("data", {}) + assert data["count"] == 1 + assert data["memberships"][0]["active"] is False + + +class TestAdminForceDeleteMembership: + """Test admin force-delete of network access requests.""" + def test_force_delete_non_admin_negative( self, integration_client,