fix(graphql): reject at query times before branch creation

backend/infrahub/api/dependencies.py

@@ -81,7 +81,11 @@ async def get_branch_params(
     branch = await registry.get_branch(db=db, branch=branch_name)
     request.state.branch_name = branch.name
 
-    return BranchParams(branch=branch, at=Timestamp(at))
+    at_ts = Timestamp(at)
+    if at is not None:
+        branch.validate_query_time(at_ts)
+
+    return BranchParams(branch=branch, at=at_ts)
 
 
 async def get_branch_dep(

backend/infrahub/core/branch/models.py

@@ -96,6 +96,27 @@ def get_created_at(self) -> str:
             raise RuntimeError(f"created_at not set for branch {self.name}")
         return self.created_at
 
+    def validate_query_time(self, at: Timestamp) -> None:
+        """Validate that `at` falls within this branch's effective lifetime.
+
+        Raises:
+            ValidationError: When `at` is earlier than the branch's effective creation time.
+
+        """
+        if self.is_default or self.is_global or self.origin_branch == self.name:
+            boundary_branch_name = self.name
+            boundary_created_at = self.get_created_at()
+        else:
+            origin = registry.get_branch_from_registry(branch=self.origin_branch)
+            boundary_branch_name = origin.name
+            boundary_created_at = origin.get_created_at()
+
+        if at < Timestamp(boundary_created_at):
+            raise ValidationError(
+                f"Requested time '{at.to_string()}' is before "
+                f"branch '{boundary_branch_name}' was created at '{boundary_created_at}'."
+            )
+
     @property
     def is_terminal(self) -> bool:
         return self.status in TERMINAL_BRANCH_STATUSES

backend/infrahub/graphql/app.py

@@ -38,7 +38,7 @@
 from infrahub.auth import AccountSession, authentication_token
 from infrahub.core.registry import registry
 from infrahub.core.timestamp import Timestamp
-from infrahub.exceptions import BranchNotFoundError, Error, PermissionDeniedError
+from infrahub.exceptions import BranchNotFoundError, Error, PermissionDeniedError, ValidationError
 from infrahub.graphql.analyzer import InfrahubGraphQLQueryAnalyzer
 from infrahub.graphql.execution import cached_parse, execute_graphql_query
 from infrahub.graphql.initialization import GraphqlParams, prepare_graphql_params
@@ -173,7 +173,7 @@ async def _get_on_get(self, request: Request) -> Response | None:
 
         return response
 
-    async def _handle_http_request(
+    async def _handle_http_request(  # noqa: PLR0915
         self, request: Request, db: InfrahubDatabase, branch: Branch, account_session: AccountSession
     ) -> JSONResponse:
         if request.app.state.response_delay:
@@ -216,18 +216,24 @@ async def _handle_http_request(
         # if the query contains some mutation, it's not currently supported to set AT manually
         if analyzed_query.contains_mutation:
             graphql_params.context.at = Timestamp()
-        elif at and branch.schema_changed_at and Timestamp(branch.schema_changed_at) > Timestamp(at):
-            schema_branch = await registry.schema.load_schema_from_db(db=db, branch=branch, at=Timestamp(at))
-            db.add_schema(name=branch.name, schema=schema_branch)
-            analyzed_query = InfrahubGraphQLQueryAnalyzer(
-                query=query,
-                schema_branch=schema_branch,
-                query_variables=variable_values,
-                schema=graphql_params.schema,
-                operation_name=operation_name,
-                branch=branch,
-                document=cached_parse(query),
-            )
+        elif at:
+            at_ts = Timestamp(at)
+            try:
+                branch.validate_query_time(at_ts)
+            except ValidationError as exc:
+                return JSONResponse(exc.api_response(), status_code=exc.HTTP_CODE)
+            if branch.schema_changed_at and Timestamp(branch.schema_changed_at) > at_ts:
+                schema_branch = await registry.schema.load_schema_from_db(db=db, branch=branch, at=at_ts)
+                db.add_schema(name=branch.name, schema=schema_branch)
+                analyzed_query = InfrahubGraphQLQueryAnalyzer(
+                    query=query,
+                    schema_branch=schema_branch,
+                    query_variables=variable_values,
+                    schema=graphql_params.schema,
+                    operation_name=operation_name,
+                    branch=branch,
+                    document=cached_parse(query),
+                )
         impacted_models = analyzed_query.query_report.impacted_models
 
         try:

backend/tests/component/api/test_10_query.py

@@ -9,6 +9,7 @@
 from infrahub.auth import AccountSession, AuthType
 from infrahub.context import BranchContext, InfrahubContext
 from infrahub.core.initialization import create_branch
+from infrahub.core.timestamp import Timestamp
 from infrahub.groups.models import RequestGraphQLQueryGroupUpdate
 from infrahub.workflows.catalogue import GRAPHQL_QUERY_GROUP_UPDATE
 
@@ -267,6 +268,33 @@ async def test_query_endpoint_wrong_branch(
     assert response.status_code == 400
 
 
+async def test_query_endpoint_at_before_branch_creation(
+    db: InfrahubDatabase,
+    client: TestClient,
+    admin_headers: dict[str, str],
+    default_branch: Branch,
+    create_test_admin: Node,
+    car_person_data: dict[str, Node],
+) -> None:
+    at_before_creation = Timestamp("2000-01-01T00:00:00Z")
+    assert at_before_creation < Timestamp(default_branch.get_created_at()), (
+        "Test precondition: the chosen `at` must be earlier than the branch's created_at"
+    )
+
+    with client:
+        response = client.get(
+            f"/api/query/query01?at={at_before_creation.to_string()}",
+            headers=admin_headers,
+        )
+
+    expected_message = (
+        f"Requested time '{at_before_creation.to_string()}' is before "
+        f"branch '{default_branch.name}' was created at '{default_branch.get_created_at()}'."
+    )
+    assert response.status_code == 422
+    assert response.json()["errors"][0]["message"] == expected_message
+
+
 async def test_query_endpoint_missing_privs(
     db: InfrahubDatabase,
     client: TestClient,

backend/tests/component/api/test_20_graphql.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 import pytest
@@ -16,6 +17,12 @@
     from infrahub.database import InfrahubDatabase
 
 
+@dataclass
+class AtBeforeCreationCase:
+    name: str
+    query_branch_name: str | None
+
+
 async def test_graphql_endpoint(
     db: InfrahubDatabase,
     client: TestClient,
@@ -117,6 +124,72 @@ async def test_graphql_endpoint_with_timestamp(
     assert sorted(names) == ["Jane", "John"]
 
 
+@pytest.mark.parametrize(
+    "case",
+    [
+        pytest.param(c, id=c.name)
+        for c in [
+            AtBeforeCreationCase(name="default_branch", query_branch_name=None),
+            AtBeforeCreationCase(name="user_branch_origin_main", query_branch_name="user-branch"),
+        ]
+    ],
+)
+async def test_graphql_endpoint_at_before_branch_creation(
+    case: AtBeforeCreationCase,
+    db: InfrahubDatabase,
+    client: TestClient,
+    admin_headers: dict[str, str],
+    default_branch: Branch,
+    create_test_admin: Node,
+    car_person_data: dict[str, Node],
+) -> None:
+    """Querying with an `at` earlier than the branch's effective lifetime must produce a clear, user-facing error.
+
+    The error always references the floor branch — the default branch itself for default-branch queries, and the
+    origin branch (`main`) for user-branch queries — never the user branch's own `created_at` or `branched_from`.
+    """
+    if case.query_branch_name is not None:
+        user_branch = await create_branch(branch_name=case.query_branch_name, db=db)
+        assert user_branch.get_created_at() != default_branch.get_created_at(), (
+            "Test precondition: the user branch must be created at a distinct time from its origin "
+            "so the boundary lookup picks a different value in each parametrized case"
+        )
+
+    at_before_creation = Timestamp("2000-01-01T00:00:00Z")
+    assert at_before_creation < Timestamp(default_branch.get_created_at()), (
+        "Test precondition: the chosen `at` must be earlier than the (origin) branch's created_at"
+    )
+
+    query = """
+    query {
+        TestPerson {
+            edges {
+                node {
+                    name {
+                        value
+                    }
+                }
+            }
+        }
+    }
+    """
+
+    url_branch_suffix = f"/{case.query_branch_name}" if case.query_branch_name else ""
+    with client:
+        response = client.post(
+            f"/graphql{url_branch_suffix}?at={at_before_creation.to_string()}",
+            json={"query": query},
+            headers=admin_headers,
+        )
+
+    expected_message = (
+        f"Requested time '{at_before_creation.to_string()}' is before "
+        f"branch '{default_branch.name}' was created at '{default_branch.get_created_at()}'."
+    )
+    assert response.status_code == 422
+    assert response.json()["errors"][0]["message"] == expected_message
+
+
 @pytest.mark.xfail(reason="Need to investigate, Currently working alone but failing when it's part of the test suite")
 async def test_graphql_endpoint_generics(
     db: InfrahubDatabase,

changelog/4076.fixed.md

@@ -0,0 +1 @@
+Fixed the unclear error message shown when picking a time earlier than the Infrahub instance was created.