Matomo

backend/src/xfd_django/xfd_api/api_methods/domain.py

@@ -10,7 +10,7 @@
 from fastapi import HTTPException
 from xfd_mini_dl.models import Domain, Service
 
-from ..auth import get_org_memberships, is_global_view_admin
+from ..auth import get_org_memberships, is_analytics_user, is_global_view_admin
 from ..helpers.filter_helpers import apply_domain_filters, sort_direction
 from ..helpers.s3_client import S3Client
 from ..schema_models.domain import DomainSearch
@@ -98,7 +98,7 @@ def search_domains(domain_search: DomainSearch, current_user):
         )
 
         # Apply global user permission filters
-        if not is_global_view_admin(current_user):
+        if not is_global_view_admin(current_user) | is_analytics_user(current_user):
             orgs = get_org_memberships(current_user)
             if not orgs:
                 # No organization memberships, return empty result

backend/src/xfd_django/xfd_api/api_methods/organization.py

@@ -13,6 +13,7 @@
 
 from ..auth import (
     get_org_memberships,
+    is_analytics_user,
     is_global_view_admin,
     is_global_write_admin,
     is_org_admin,
@@ -42,14 +43,18 @@ def list_organizations(current_user):
     """List organizations that the user is a member of or has access to."""
     try:
         # Check if user is GlobalViewAdmin or has memberships
-        if not is_global_view_admin(current_user) and not get_org_memberships(
-            current_user
+        if (
+            not is_global_view_admin(current_user)
+            and not is_analytics_user(current_user)
+            and not get_org_memberships(current_user)
         ):
             return []
 
         # Define filter for organizations based on admin status
         org_filter = {}
-        if not is_global_view_admin(current_user):
+        if not is_global_view_admin(current_user) and not is_analytics_user(
+            current_user
+        ):
             org_filter["id__in"] = get_org_memberships(current_user)
         org_filter["parent"] = None
 
@@ -128,9 +133,10 @@ def get_organization(organization_id, current_user):
     try:
         # Authorization checks
         if not (
+            is_analytics_user(current_user),
             is_org_admin(current_user, organization_id)
             or is_global_view_admin(current_user)
-            or is_regional_admin_for_organization(current_user, organization_id)
+            or is_regional_admin_for_organization(current_user, organization_id),
         ):
             raise HTTPException(status_code=403, detail="Unauthorized")
 
@@ -337,8 +343,10 @@ def get_all_regions(current_user):
     """Get all regions."""
     try:
         # Check if user is GlobalViewAdmin or has memberships
-        if not is_global_view_admin(current_user) and not get_org_memberships(
-            current_user
+        if (
+            not is_global_view_admin(current_user)
+            and not is_analytics_user(current_user)
+            and not get_org_memberships(current_user)
         ):
             raise HTTPException(status_code=403, detail="Unauthorized")
 
@@ -997,15 +1005,19 @@ def list_organizations_v2(state, region_id, current_user):
     """List organizations that the user is a member of or has access to."""
     try:
         # Check if user is GlobalViewAdmin or has memberships
-        if not is_global_view_admin(current_user) and not get_org_memberships(
-            current_user
+        if (
+            not is_global_view_admin(current_user)
+            and not is_analytics_user(current_user)
+            and not get_org_memberships(current_user)
         ):
             return []
 
         # Prepare the filter criteria
         filter_criteria = Q()
 
-        if not is_global_view_admin(current_user):
+        if not is_global_view_admin(current_user) and not is_analytics_user(
+            current_user
+        ):
             filter_criteria &= Q(id__in=get_org_memberships(current_user))
 
         if state:
@@ -1066,8 +1078,10 @@ def search_organizations_task(search_body, current_user: User):
     """Handle the logic for searching organizations in Elasticsearch."""
     try:
         # Check if user is GlobalViewAdmin or has memberships
-        if not is_global_view_admin(current_user) and not get_org_memberships(
-            current_user
+        if (
+            not is_global_view_admin(current_user)
+            and not is_analytics_user(current_user)
+            and not get_org_memberships(current_user)
         ):
             return []
 

backend/src/xfd_django/xfd_api/api_methods/proxy.py

@@ -5,51 +5,70 @@
 
 # Third-Party Libraries
 from fastapi import Request
-from fastapi.responses import Response
+from fastapi.responses import RedirectResponse, Response
 import httpx
 
 
-# Helper function to handle cookie manipulation
-def manipulate_cookie(request: Request, cookie_name: str):
-    """Manipulate cookie."""
-    cookies = request.cookies.get(cookie_name)
-    if cookies:
-        return {cookie_name: cookies}
-    return {}
-
-
 # Helper function to proxy requests
 async def proxy_request(
     request: Request,
     target_url: str,
     path: Optional[str] = None,
     cookie_name: Optional[str] = None,
 ):
-    """Proxy the request to the target URL."""
+    """Proxy requests to the specified target URL with optional cookie handling."""
+    print("Proxying request to target URL: {}".format(target_url))
     headers = dict(request.headers)
 
-    # Cookie manipulation for specific cookie names
+    # Include specified cookie in the headers if present
     if cookie_name:
-        cookies = manipulate_cookie(request, cookie_name)
+        print("Cookie name: {}".format(cookie_name))
+        cookies = request.cookies.get(cookie_name)
         if cookies:
-            headers["Cookie"] = "{}={}".format(cookie_name, cookies[cookie_name])
+            headers["Cookie"] = "{}={}".format(cookie_name, cookies)
 
-    # Make the request to the target URL
-    async with httpx.AsyncClient() as client:
+    # Send the request to the target
+    async with httpx.AsyncClient(timeout=httpx.Timeout(90.0)) as client:
         proxy_response = await client.request(
             method=request.method,
             url="{}/{}".format(target_url, path),
             headers=headers,
             params=request.query_params,
             content=await request.body(),
         )
-
-    # Remove chunked encoding for API Gateway compatibility
+    # Adjust response headers
     proxy_response_headers = dict(proxy_response.headers)
-    proxy_response_headers.pop("transfer-encoding", None)
+    for header in ["content-encoding", "transfer-encoding", "content-length"]:
+        proxy_response_headers.pop(header, None)
 
     return Response(
         content=proxy_response.content,
         status_code=proxy_response.status_code,
         headers=proxy_response_headers,
     )
+
+
+async def matomo_proxy_handler(
+    request: Request,
+    path: str,
+    MATOMO_URL: str,
+):
+    """
+    Handle Matomo-specific proxy logic.
+
+    Includes public paths, font redirects, and authentication for private paths.
+    """
+    # Redirect font requests to CDN
+    font_paths = {
+        "/plugins/Morpheus/fonts/matomo.woff2": "https://cdn.jsdelivr.net/gh/matomo-org/[email protected]/plugins/Morpheus/fonts/matomo.woff2",
+        "/plugins/Morpheus/fonts/matomo.woff": "https://cdn.jsdelivr.net/gh/matomo-org/[email protected]/plugins/Morpheus/fonts/matomo.woff",
+        "/plugins/Morpheus/fonts/matomo.ttf": "https://cdn.jsdelivr.net/gh/matomo-org/[email protected]/plugins/Morpheus/fonts/matomo.ttf",
+    }
+    if path in font_paths:
+        return RedirectResponse(url=font_paths[path])
+
+    return await proxy_request(
+        request=request,
+        target_url=MATOMO_URL,
+        path=path,
+    )

backend/src/xfd_django/xfd_api/api_methods/search.py

@@ -9,6 +9,7 @@
 from xfd_api.auth import (
     get_org_memberships,
     get_tag_organizations,
+    is_analytics_user,
     is_global_view_admin,
 )
 from xfd_api.helpers.elastic_search import build_request
@@ -22,7 +23,7 @@ async def get_options(search_body, user) -> Dict[str, Any]:
     """Get Elastic Search options."""
     if search_body.organization_id and (
         search_body.organization_id in get_org_memberships(user)
-        or is_global_view_admin(user)
+        or (is_global_view_admin(user) | is_analytics_user(user))
     ):
         return {
             "organization_ids": [search_body.organization_id],
@@ -36,7 +37,7 @@ async def get_options(search_body, user) -> Dict[str, Any]:
 
     return {
         "organization_ids": get_org_memberships(user),
-        "match_all_organizations": is_global_view_admin(user),
+        "match_all_organizations": is_global_view_admin(user) | is_analytics_user(user),
     }
 
 

backend/src/xfd_django/xfd_api/api_methods/stats.py

@@ -21,7 +21,7 @@
     VulnScanSummary,
 )
 
-from ..auth import get_org_memberships, is_global_view_admin
+from ..auth import get_org_memberships, is_analytics_user, is_global_view_admin
 
 
 # GET: /stats
@@ -109,7 +109,8 @@ async def safe_fetch(fetch_fn, *args, **kwargs):
         }
     except Exception as e:
         raise HTTPException(
-            status_code=500, detail="An unexpected error occurred: {}".format(e)
+            status_code=500,
+            detail="An unexpected error occurred: {}".format(e),
         )
 
 
@@ -453,7 +454,7 @@ def get_vs_condensed_trending_data(filters, current_user):
         raise HTTPException(status_code=404, detail="Organization not found.")
 
     if (
-        not is_global_view_admin(current_user)
+        not is_global_view_admin(current_user) | is_analytics_user(current_user)
         and not current_user.user_type == "regionalAdmin"
     ):
         org_ids = get_org_memberships(current_user)
@@ -538,6 +539,7 @@ def get_vs_trending_data(filters, current_user):
 
     if (
         not is_global_view_admin(current_user)
+        and not is_analytics_user(current_user)
         and not current_user.user_type == "regionalAdmin"
     ):
         org_ids = get_org_memberships(current_user)

backend/src/xfd_django/xfd_api/api_methods/user.py

@@ -13,6 +13,7 @@
 
 from ..auth import (
     can_access_user,
+    is_analytics_user,
     is_global_view_admin,
     is_global_write_admin,
     is_org_admin,
@@ -193,7 +194,11 @@ def get_users(current_user):
     """Retrieve a list of all users."""
     try:
         # Check if user is a regional admin or global admin
-        if not is_global_view_admin(current_user) | is_regional_admin(current_user):
+        if (
+            not is_global_view_admin(current_user)
+            | is_regional_admin(current_user)
+            | is_analytics_user(current_user)
+        ):
             raise HTTPException(status_code=401, detail="Unauthorized")
 
         users = User.objects.all().prefetch_related("roles__organization")
@@ -241,7 +246,7 @@ def get_users(current_user):
 def get_users_by_region_id(region_id, current_user):
     """List users with specific region_id."""
     try:
-        if not is_regional_admin(current_user):
+        if not is_regional_admin(current_user) | is_analytics_user(current_user):
             raise HTTPException(status_code=401, detail="Unauthorized")
 
         if not region_id:
@@ -357,7 +362,11 @@ def get_users_v2(state, region_id, invite_pending, current_user):
     """Retrieve a list of users based on optional filter parameters."""
     try:
         # Check if user is a regional admin or global admin
-        if not is_regional_admin(current_user) | is_global_view_admin(current_user):
+        if (
+            not is_regional_admin(current_user)
+            | is_global_view_admin(current_user)
+            | is_analytics_user(current_user)
+        ):
             raise HTTPException(status_code=401, detail="Unauthorized")
 
         filters = {}

backend/src/xfd_django/xfd_api/api_methods/vulnerability.py

@@ -23,7 +23,7 @@
     Vulnerability,
 )
 
-from ..auth import get_org_memberships, is_global_view_admin
+from ..auth import get_org_memberships, is_analytics_user, is_global_view_admin
 from ..helpers.filter_helpers import apply_vuln_filters, sort_direction
 from ..helpers.s3_client import S3Client
 from ..models import Domain
@@ -105,6 +105,7 @@ def get_vulnerability_by_scan_source_and_id(
 
         if (
             not is_global_view_admin(current_user)
+            and not is_analytics_user(current_user)
             and not current_user.user_type == "regionalAdmin"
         ):
             org_ids = get_org_memberships(current_user)
@@ -385,6 +386,7 @@ def search_vulnerabilities(vulnerability_search: VulnerabilitySearch, current_us
         # Permissions check
         if (
             not is_global_view_admin(current_user)
+            and not is_analytics_user(current_user)
             and not current_user.user_type == "regionalAdmin"
         ):
             org_ids = get_org_memberships(current_user)

backend/src/xfd_django/xfd_api/auth.py

@@ -338,6 +338,11 @@ def is_regional_admin(current_user) -> bool:
     return current_user and current_user.user_type in ["regionalAdmin", "globalAdmin"]
 
 
+def is_analytics_user(current_user) -> bool:
+    """Check if the user has analytics permissions."""
+    return current_user and current_user.user_type in ["analytics", "globalAdmin"]
+
+
 def is_org_admin(current_user, organization_id) -> bool:
     """Check if the user is an admin of the given organization."""
     if not organization_id:
@@ -461,6 +466,7 @@ def get_stats_org_ids(current_user, filters):
                 or (is_regional_admin_for_organization(current_user, org_id))
                 or (is_org_admin(current_user, org_id))
                 or (get_org_memberships(current_user))
+                or (is_analytics_user(current_user))
             ):
                 organization_ids.add(org_id)
 
@@ -483,11 +489,21 @@ def get_stats_org_ids(current_user, filters):
         for tag_id in tags_filter:
             organizations_by_tag = get_tag_organizations(current_user, tag_id)
             organization_ids.update(organizations_by_tag)
-
-    # Case 3: Regional admin
+    # Case 3: Analytics view
+    elif is_analytics_user(current_user):
+        # Get organizations by region
+        if regions_filter:
+            organizations_by_region = Organization.objects.filter(
+                region_id__in=regions_filter
+            ).values_list("id", flat=True)
+            organization_ids.update(organizations_by_region)
+        # Get organizations by tag
+        for tag_id in tags_filter:
+            organizations_by_tag = get_tag_organizations(current_user, tag_id)
+            organization_ids.update(organizations_by_tag)
+    # Case 4: Regional admin
     elif current_user.user_type in ["regionalAdmin"]:
         user_region_id = current_user.region_id
-
         # Allow only organizations in the user's region
         organizations_in_region = Organization.objects.filter(
             region_id=user_region_id
@@ -508,7 +524,7 @@ def get_stats_org_ids(current_user, filters):
             ]
             organization_ids.update(regional_tag_organizations)
 
-    # Case 4: Standard user
+    # Case 5: Standard user
     else:
         # Allow only organizations where the user is a member
         user_organization_ids = current_user.roles.values_list(

backend/src/xfd_django/xfd_api/models.py

@@ -533,6 +533,7 @@ class Meta:
 class UserType(models.TextChoices):
     """User type definition."""
 
+    ANALYTICS = "analytics"
     GLOBAL_ADMIN = "globalAdmin"
     GLOBAL_VIEW = "globalView"
     REGIONAL_ADMIN = "regionalAdmin"

backend/src/xfd_django/xfd_api/schema_models/user.py

@@ -16,6 +16,7 @@
 class UserType(Enum):
     """User Type."""
 
+    ANALYTICS = "analytics"
     GLOBAL_ADMIN = "globalAdmin"
     GLOBAL_VIEW = "globalView"
     REGIONAL_ADMIN = "regionalAdmin"