diff --git a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/roles.py b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/roles.py
index 57c3cc31668ee..2329d679ebf35 100644
--- a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/roles.py
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/roles.py
@@ -40,6 +40,12 @@ class ActionResourceResponse(BaseModel):
     resource: ResourceResponse
 
 
+class Role(BaseModel):
+    """Lightweight role reference used by /users schemas."""
+
+    name: str
+
+
 class RoleBody(StrictBaseModel):
     """Incoming payload for creating/updating a role."""
 
diff --git a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/users.py b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/users.py
new file mode 100644
index 0000000000000..8cc1b1387eab2
--- /dev/null
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/datamodels/users.py
@@ -0,0 +1,62 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from pydantic import Field, SecretStr, field_validator
+
+from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import Role
+
+if TYPE_CHECKING:
+    from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import Role
+
+
+class UserBody(StrictBaseModel):
+    """Incoming payload for creating a user."""
+
+    username: str = Field(min_length=1)
+    email: str = Field(min_length=1)
+    first_name: str = Field(min_length=1)
+    last_name: str = Field(min_length=1)
+    roles: list[Role] | None = None
+    password: SecretStr
+
+
+class UserResponse(BaseModel):
+    """Outgoing representation of a user (no password)."""
+
+    username: str
+    email: str
+    first_name: str
+    last_name: str
+    roles: list[Role] | None = None
+    active: bool | None = None
+    last_login: datetime | None = None
+    login_count: int | None = None
+    fail_login_count: int | None = None
+    created_on: datetime | None = None
+    changed_on: datetime | None = None
+
+    @field_validator("created_on", "changed_on")
+    @classmethod
+    def _coerce_tzaware(cls, v: datetime | None) -> datetime | None:
+        if v is None:
+            return None
+        return v if (v.tzinfo and v.utcoffset() is not None) else v.replace(tzinfo=timezone.utc)
diff --git a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/openapi/v2-fab-auth-manager-generated.yaml b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/openapi/v2-fab-auth-manager-generated.yaml
index f3f7fbfe90b86..18702657693ac 100644
--- a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/openapi/v2-fab-auth-manager-generated.yaml
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/openapi/v2-fab-auth-manager-generated.yaml
@@ -324,6 +324,64 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /auth/fab/v1/users:
+    post:
+      tags:
+      - FabAuthManager
+      summary: Create User
+      operationId: create_user
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/UserBody'
+        required: true
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/UserResponse'
+        '400':
+          description: Bad Request
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '401':
+          description: Unauthorized
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '403':
+          description: Forbidden
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '409':
+          description: Conflict
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '500':
+          description: Internal Server Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
 components:
   schemas:
     ActionResourceResponse:
@@ -404,6 +462,16 @@ components:
       - name
       title: ResourceResponse
       description: Outgoing representation of a resource.
+    Role:
+      properties:
+        name:
+          type: string
+          title: Name
+      type: object
+      required:
+      - name
+      title: Role
+      description: Lightweight role reference used by /users schemas.
     RoleBody:
       properties:
         name:
@@ -452,6 +520,108 @@ components:
       - name
       title: RoleResponse
       description: Outgoing representation of a role and its permissions.
+    UserBody:
+      properties:
+        username:
+          type: string
+          minLength: 1
+          title: Username
+        email:
+          type: string
+          minLength: 1
+          title: Email
+        first_name:
+          type: string
+          minLength: 1
+          title: First Name
+        last_name:
+          type: string
+          minLength: 1
+          title: Last Name
+        roles:
+          anyOf:
+          - items:
+              $ref: '#/components/schemas/Role'
+            type: array
+          - type: 'null'
+          title: Roles
+        password:
+          type: string
+          format: password
+          title: Password
+          writeOnly: true
+      additionalProperties: false
+      type: object
+      required:
+      - username
+      - email
+      - first_name
+      - last_name
+      - password
+      title: UserBody
+      description: Incoming payload for creating a user.
+    UserResponse:
+      properties:
+        username:
+          type: string
+          title: Username
+        email:
+          type: string
+          title: Email
+        first_name:
+          type: string
+          title: First Name
+        last_name:
+          type: string
+          title: Last Name
+        roles:
+          anyOf:
+          - items:
+              $ref: '#/components/schemas/Role'
+            type: array
+          - type: 'null'
+          title: Roles
+        active:
+          anyOf:
+          - type: boolean
+          - type: 'null'
+          title: Active
+        last_login:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Last Login
+        login_count:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Login Count
+        fail_login_count:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Fail Login Count
+        created_on:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Created On
+        changed_on:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Changed On
+      type: object
+      required:
+      - username
+      - email
+      - first_name
+      - last_name
+      title: UserResponse
+      description: Outgoing representation of a user (no password).
     ValidationError:
       properties:
         loc:
diff --git a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/routes/users.py b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/routes/users.py
new file mode 100644
index 0000000000000..c76608f00e496
--- /dev/null
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/routes/users.py
@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastapi import Depends, status
+
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.users import UserBody, UserResponse
+from airflow.providers.fab.auth_manager.api_fastapi.security import requires_fab_custom_view
+from airflow.providers.fab.auth_manager.api_fastapi.services.users import FABAuthManagerUsers
+from airflow.providers.fab.auth_manager.cli_commands.utils import get_application_builder
+from airflow.providers.fab.www.security import permissions
+
+if TYPE_CHECKING:
+    from airflow.providers.fab.auth_manager.api_fastapi.datamodels.users import UserBody, UserResponse
+
+users_router = AirflowRouter(prefix="/fab/v1", tags=["FabAuthManager"])
+
+
+@users_router.post(
+    "/users",
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_400_BAD_REQUEST,
+            status.HTTP_401_UNAUTHORIZED,
+            status.HTTP_403_FORBIDDEN,
+            status.HTTP_409_CONFLICT,
+            status.HTTP_500_INTERNAL_SERVER_ERROR,
+        ]
+    ),
+    dependencies=[Depends(requires_fab_custom_view("POST", permissions.RESOURCE_USER))],
+)
+def create_user(body: UserBody) -> UserResponse:
+    with get_application_builder():
+        return FABAuthManagerUsers.create_user(body=body)
diff --git a/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/services/users.py b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/services/users.py
new file mode 100644
index 0000000000000..6e1ed9f4f280c
--- /dev/null
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/api_fastapi/services/users.py
@@ -0,0 +1,96 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastapi import HTTPException, status
+
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import Role
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.users import UserBody, UserResponse
+from airflow.providers.fab.www.utils import get_fab_auth_manager
+
+if TYPE_CHECKING:
+    from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import Role
+    from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
+
+
+class FABAuthManagerUsers:
+    """Service layer for FAB Auth Manager user operations (create, validate, sync)."""
+
+    @staticmethod
+    def _resolve_roles(
+        sm: FabAirflowSecurityManagerOverride, role_refs: list[Role] | None
+    ) -> tuple[list, list[str]]:
+        seen = set()
+        roles: list = []
+        missing: list[str] = []
+        for r in role_refs or []:
+            if r.name in seen:
+                continue
+            seen.add(r.name)
+            role = sm.find_role(r.name)
+            (roles if role else missing).append(role or r.name)
+        return roles, missing
+
+    @classmethod
+    def create_user(cls, body: UserBody) -> UserResponse:
+        security_manager = get_fab_auth_manager().security_manager
+
+        existing_username = security_manager.find_user(username=body.username)
+        if existing_username:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT,
+                detail=f"Username `{body.username}` already exists. Use PATCH to update.",
+            )
+
+        existing_email = security_manager.find_user(email=body.email)
+        if existing_email:
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail=f"The email `{body.email}` is already taken."
+            )
+
+        roles_to_add, missing_role_names = cls._resolve_roles(security_manager, body.roles)
+        if missing_role_names:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Unknown roles: {', '.join(repr(n) for n in missing_role_names)}",
+            )
+        if not roles_to_add:
+            default_role = security_manager.find_role(security_manager.auth_user_registration_role)
+            if default_role is None:
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail="Default registration role is not configured or not found.",
+                )
+            roles_to_add.append(default_role)
+
+        created = security_manager.add_user(
+            username=body.username,
+            email=body.email,
+            first_name=body.first_name,
+            last_name=body.last_name,
+            role=roles_to_add,
+            password=body.password.get_secret_value(),
+        )
+        if not created:
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail=f"Failed to add user `{body.username}`",
+            )
+
+        return UserResponse.model_validate(created)
diff --git a/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py b/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
index eaf98737394f0..8c584921c0c33 100644
--- a/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
+++ b/providers/fab/src/airflow/providers/fab/auth_manager/fab_auth_manager.py
@@ -227,6 +227,7 @@ def get_fastapi_app(self) -> FastAPI | None:
             login_router,
         )
         from airflow.providers.fab.auth_manager.api_fastapi.routes.roles import roles_router
+        from airflow.providers.fab.auth_manager.api_fastapi.routes.users import users_router
 
         flask_app = create_app(enable_plugins=False)
 
@@ -243,6 +244,7 @@ def get_fastapi_app(self) -> FastAPI | None:
         # Add the login router to the FastAPI app
         app.include_router(login_router)
         app.include_router(roles_router)
+        app.include_router(users_router)
 
         app.mount("/", WSGIMiddleware(flask_app))
 
diff --git a/providers/fab/tests/unit/fab/auth_manager/api_fastapi/datamodels/test_users.py b/providers/fab/tests/unit/fab/auth_manager/api_fastapi/datamodels/test_users.py
new file mode 100644
index 0000000000000..8a4d9a2ba87c2
--- /dev/null
+++ b/providers/fab/tests/unit/fab/auth_manager/api_fastapi/datamodels/test_users.py
@@ -0,0 +1,145 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+import types
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from pydantic import SecretStr, ValidationError
+
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import Role
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.users import (
+    UserBody,
+    UserResponse,
+)
+
+
+class TestUserModels:
+    def test_userbody_accepts_role_dicts_and_Roles_and_secretstr(self):
+        data = {
+            "username": "alice",
+            "email": "alice@example.com",
+            "first_name": "Alice",
+            "last_name": "Liddell",
+            "password": "s3cr3t",  # should coerce into SecretStr
+            "roles": [{"name": "Admin"}, Role(name="User")],
+        }
+        body = UserBody.model_validate(data)
+        assert body.username == "alice"
+        assert body.email == "alice@example.com"
+        assert isinstance(body.password, SecretStr)
+        assert body.password.get_secret_value() == "s3cr3t"
+        assert body.roles is not None
+        assert [r.name for r in body.roles] == ["Admin", "User"]
+
+        # SecretStr should be masked on JSON serialization by default
+        dumped_json = body.model_dump_json()
+        payload = json.loads(dumped_json)
+        assert payload["password"] == "**********"
+
+    def test_userbody_roles_default_none_when_omitted(self):
+        body = UserBody.model_validate(
+            {
+                "username": "bob",
+                "email": "bob@example.com",
+                "first_name": "Bob",
+                "last_name": "Builder",
+                "password": "pw",
+            }
+        )
+        assert body.roles is None
+
+    @pytest.mark.parametrize(
+        "patch",
+        [
+            {"username": ""},  # min_length=1
+            {"email": ""},  # min_length=1
+            {"first_name": ""},  # min_length=1
+            {"last_name": ""},  # min_length=1
+        ],
+    )
+    def test_userbody_min_length_enforced(self, patch):
+        base = {
+            "username": "ok",
+            "email": "ok@example.com",
+            "first_name": "OK",
+            "last_name": "User",
+            "password": "pw",
+        }
+        base.update(patch)
+        with pytest.raises(ValidationError):
+            UserBody.model_validate(base)
+
+    def test_userbody_password_is_required(self):
+        with pytest.raises(ValidationError):
+            UserBody.model_validate(
+                {
+                    "username": "no-pass",
+                    "email": "np@example.com",
+                    "first_name": "No",
+                    "last_name": "Pass",
+                }
+            )
+
+    def test_userresponse_coerces_naive_datetimes_to_utc(self):
+        naive_created = datetime(2025, 1, 2, 3, 4, 5)
+        resp = UserResponse.model_validate(
+            {
+                "username": "alice",
+                "email": "alice@example.com",
+                "first_name": "Alice",
+                "last_name": "Liddell",
+                "created_on": naive_created,
+            }
+        )
+        assert resp.created_on is not None
+        assert resp.created_on.tzinfo is timezone.utc
+        assert resp.created_on.utcoffset() == timedelta(0)
+        assert resp.created_on.replace(tzinfo=None) == naive_created
+
+    def test_userresponse_preserves_aware_datetimes(self):
+        aware = datetime(2024, 12, 1, 9, 30, tzinfo=timezone(timedelta(hours=9)))
+        resp = UserResponse.model_validate(
+            {
+                "username": "bob",
+                "email": "bob@example.com",
+                "first_name": "Bob",
+                "last_name": "Builder",
+                "changed_on": aware,
+            }
+        )
+        assert resp.changed_on == aware
+        assert resp.changed_on.tzinfo == aware.tzinfo
+
+    def test_userresponse_model_validate_from_simple_namespace(self):
+        obj = types.SimpleNamespace(
+            username="eve",
+            email="eve@example.com",
+            first_name="Eve",
+            last_name="Adams",
+            roles=[types.SimpleNamespace(name="Viewer")],
+            active=True,
+            login_count=10,
+        )
+        resp = UserResponse.model_validate(obj)
+        assert resp.username == "eve"
+        assert resp.roles is not None
+        assert resp.roles[0].name == "Viewer"
+        assert resp.active is True
+        assert resp.login_count == 10
diff --git a/providers/fab/tests/unit/fab/auth_manager/api_fastapi/routes/test_users.py b/providers/fab/tests/unit/fab/auth_manager/api_fastapi/routes/test_users.py
new file mode 100644
index 0000000000000..2ac1597cb9ef8
--- /dev/null
+++ b/providers/fab/tests/unit/fab/auth_manager/api_fastapi/routes/test_users.py
@@ -0,0 +1,173 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from contextlib import nullcontext as _noop_cm
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.users import UserResponse
+
+
+@pytest.mark.db_test
+class TestUsers:
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.users.FABAuthManagerUsers")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.users.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_create_user_ok(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_users, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = True
+        mock_get_auth_manager.return_value = mgr
+
+        dummy_out = UserResponse(
+            username="alice",
+            email="alice@example.com",
+            first_name="Alice",
+            last_name="Liddell",
+            roles=None,
+            active=True,
+            login_count=0,
+            fail_login_count=0,
+        )
+        mock_users.create_user.return_value = dummy_out
+
+        with as_user():
+            payload = {
+                "username": "alice",
+                "email": "alice@example.com",
+                "first_name": "Alice",
+                "last_name": "Liddell",
+                "password": "s3cr3t",
+                "roles": [{"name": "Viewer"}],
+            }
+            resp = test_client.post("/fab/v1/users", json=payload)
+            assert resp.status_code == 200
+            assert resp.json() == dummy_out.model_dump(by_alias=True)
+            mock_users.create_user.assert_called_once()
+
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.users.FABAuthManagerUsers")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.users.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_create_user_forbidden(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_users, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = False
+        mock_get_auth_manager.return_value = mgr
+
+        with as_user():
+            resp = test_client.post(
+                "/fab/v1/users",
+                json={
+                    "username": "bob",
+                    "email": "bob@example.com",
+                    "first_name": "Bob",
+                    "last_name": "Builder",
+                    "password": "pw",
+                },
+            )
+            assert resp.status_code == 403
+            mock_users.create_user.assert_not_called()
+
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.users.FABAuthManagerUsers")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.users.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_create_user_validation_422_empty_username(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_users, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = True
+        mock_get_auth_manager.return_value = mgr
+
+        with as_user():
+            resp = test_client.post(
+                "/fab/v1/users",
+                json={
+                    "username": "",
+                    "email": "e@example.com",
+                    "first_name": "E",
+                    "last_name": "Mpty",
+                    "password": "pw",
+                },
+            )
+            assert resp.status_code == 422
+            mock_users.create_user.assert_not_called()
+
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.users.FABAuthManagerUsers")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.users.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_create_user_validation_422_missing_username(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_users, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = True
+        mock_get_auth_manager.return_value = mgr
+
+        with as_user():
+            resp = test_client.post(
+                "/fab/v1/users",
+                json={
+                    "email": "nouser@example.com",
+                    "first_name": "No",
+                    "last_name": "User",
+                    "password": "pw",
+                },
+            )
+            assert resp.status_code == 422
+            mock_users.create_user.assert_not_called()
+
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.routes.users.FABAuthManagerUsers")
+    @patch("airflow.providers.fab.auth_manager.api_fastapi.security.get_auth_manager")
+    @patch(
+        "airflow.providers.fab.auth_manager.api_fastapi.routes.users.get_application_builder",
+        return_value=_noop_cm(),
+    )
+    def test_create_user_validation_422_missing_password(
+        self, mock_get_application_builder, mock_get_auth_manager, mock_users, test_client, as_user
+    ):
+        mgr = MagicMock()
+        mgr.is_authorized_custom_view.return_value = True
+        mock_get_auth_manager.return_value = mgr
+
+        with as_user():
+            resp = test_client.post(
+                "/fab/v1/users",
+                json={
+                    "username": "no-pass",
+                    "email": "np@example.com",
+                    "first_name": "No",
+                    "last_name": "Pass",
+                    # password missing
+                },
+            )
+            assert resp.status_code == 422
+            mock_users.create_user.assert_not_called()
diff --git a/providers/fab/tests/unit/fab/auth_manager/api_fastapi/services/test_users.py b/providers/fab/tests/unit/fab/auth_manager/api_fastapi/services/test_users.py
new file mode 100644
index 0000000000000..759f98959b4e1
--- /dev/null
+++ b/providers/fab/tests/unit/fab/auth_manager/api_fastapi/services/test_users.py
@@ -0,0 +1,206 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import types
+from unittest.mock import MagicMock, patch
+
+import pytest
+from fastapi import HTTPException
+
+from airflow.providers.fab.auth_manager.api_fastapi.datamodels.roles import Role
+from airflow.providers.fab.auth_manager.api_fastapi.services.users import FABAuthManagerUsers
+
+
+@pytest.fixture
+def fab_auth_manager():
+    return MagicMock()
+
+
+@pytest.fixture
+def security_manager():
+    sm = MagicMock()
+
+    def _find_role(name: str):
+        if name in {"Admin", "User"}:
+            return types.SimpleNamespace(name=name)
+        return None
+
+    sm.find_role.side_effect = _find_role
+    sm.auth_user_registration_role = "User"
+    return sm
+
+
+def _make_user_obj(
+    *,
+    username: str,
+    email: str,
+    first_name: str,
+    last_name: str,
+    roles: list[str] | None = None,
+):
+    role_objs = [types.SimpleNamespace(name=r) for r in (roles or [])]
+    return types.SimpleNamespace(
+        username=username,
+        email=email,
+        first_name=first_name,
+        last_name=last_name,
+        roles=role_objs or None,
+        active=True,
+        login_count=0,
+    )
+
+
+@patch("airflow.providers.fab.auth_manager.api_fastapi.services.users.get_fab_auth_manager")
+class TestUsersService:
+    def setup_method(self):
+        self.password_mock = MagicMock()
+        self.password_mock.get_secret_value.return_value = "pw"
+
+        self.body_base = types.SimpleNamespace(
+            username="alice",
+            email="alice@example.com",
+            first_name="Alice",
+            last_name="Liddell",
+            password=self.password_mock,
+            roles=None,
+        )
+
+        self.body_with_roles_admin_dupe = types.SimpleNamespace(
+            username="bob",
+            email="bob@example.com",
+            first_name="Bob",
+            last_name="Builder",
+            password=MagicMock(get_secret_value=MagicMock(return_value="pw2")),
+            roles=[types.SimpleNamespace(name="Admin"), types.SimpleNamespace(name="Admin")],
+        )
+
+        self.body_with_missing_role = types.SimpleNamespace(
+            username="eve",
+            email="eve@example.com",
+            first_name="Eve",
+            last_name="Adams",
+            password=MagicMock(get_secret_value=MagicMock(return_value="pw3")),
+            roles=[types.SimpleNamespace(name="NOPE")],
+        )
+
+    def test_create_user_success_with_default_role(
+        self, get_fab_auth_manager, fab_auth_manager, security_manager
+    ):
+        security_manager.find_user.side_effect = [None, None]
+        security_manager.add_user.return_value = _make_user_obj(
+            username=self.body_base.username,
+            email=self.body_base.email,
+            first_name=self.body_base.first_name,
+            last_name=self.body_base.last_name,
+            roles=["User"],
+        )
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        out = FABAuthManagerUsers.create_user(self.body_base)
+
+        assert out.username == "alice"
+        assert out.email == "alice@example.com"
+        called_roles = security_manager.add_user.call_args.kwargs["role"]
+        assert len(called_roles) == 1
+        assert called_roles[0].name == "User"
+        self.password_mock.get_secret_value.assert_called_once()
+
+    def test_create_user_success_with_explicit_roles_and_dedup(
+        self, get_fab_auth_manager, fab_auth_manager, security_manager
+    ):
+        security_manager.find_user.side_effect = [None, None]
+        security_manager.add_user.return_value = _make_user_obj(
+            username=self.body_with_roles_admin_dupe.username,
+            email=self.body_with_roles_admin_dupe.email,
+            first_name=self.body_with_roles_admin_dupe.first_name,
+            last_name=self.body_with_roles_admin_dupe.last_name,
+            roles=["Admin"],
+        )
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        out = FABAuthManagerUsers.create_user(self.body_with_roles_admin_dupe)
+
+        assert out.username == "bob"
+        roles_arg = security_manager.add_user.call_args.kwargs["role"]
+        assert len(roles_arg) == 1
+        assert roles_arg[0].name == "Admin"
+
+    def test_create_user_conflict_username(self, get_fab_auth_manager, fab_auth_manager, security_manager):
+        security_manager.find_user.side_effect = [object()]
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        with pytest.raises(HTTPException) as ex:
+            FABAuthManagerUsers.create_user(self.body_base)
+        assert ex.value.status_code == 409
+        assert "Username" in ex.value.detail
+
+    def test_create_user_conflict_email(self, get_fab_auth_manager, fab_auth_manager, security_manager):
+        security_manager.find_user.side_effect = [None, object()]
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        with pytest.raises(HTTPException) as ex:
+            FABAuthManagerUsers.create_user(self.body_base)
+        assert ex.value.status_code == 409
+        assert "email" in ex.value.detail
+
+    def test_create_user_unknown_roles(self, get_fab_auth_manager, fab_auth_manager, security_manager):
+        security_manager.find_user.side_effect = [None, None]
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        with pytest.raises(HTTPException) as ex:
+            FABAuthManagerUsers.create_user(self.body_with_missing_role)
+        assert ex.value.status_code == 400
+        assert "Unknown roles" in ex.value.detail
+
+    def test_create_user_default_role_missing(self, get_fab_auth_manager, fab_auth_manager, security_manager):
+        security_manager.find_user.side_effect = [None, None]
+        security_manager.auth_user_registration_role = "MissingDefault"
+        security_manager.find_role.side_effect = lambda n: None if n == "MissingDefault" else None
+
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        with pytest.raises(HTTPException) as ex:
+            FABAuthManagerUsers.create_user(self.body_base)
+        assert ex.value.status_code == 500
+        assert "Default registration role" in ex.value.detail
+
+    def test_create_user_add_user_failed(self, get_fab_auth_manager, fab_auth_manager, security_manager):
+        security_manager.find_user.side_effect = [None, None]
+        security_manager.add_user.return_value = None
+
+        fab_auth_manager.security_manager = security_manager
+        get_fab_auth_manager.return_value = fab_auth_manager
+
+        with pytest.raises(HTTPException) as ex:
+            FABAuthManagerUsers.create_user(self.body_base)
+        assert ex.value.status_code == 500
+        assert "Failed to add user" in ex.value.detail
+
+    def test_resolve_roles_returns_found_and_missing(self, get_fab_auth_manager, security_manager):
+        found, missing = FABAuthManagerUsers._resolve_roles(
+            security_manager,
+            [Role(name="Admin"), Role(name="NOPE"), Role(name="Admin")],
+        )
+        assert [r.name for r in found] == ["Admin"]
+        assert missing == ["NOPE"]
diff --git a/scripts/in_container/install_airflow_and_providers.py b/scripts/in_container/install_airflow_and_providers.py
index 68ca7254f157a..f04c3e99c8a12 100755
--- a/scripts/in_container/install_airflow_and_providers.py
+++ b/scripts/in_container/install_airflow_and_providers.py
@@ -435,6 +435,12 @@ def find_installation_spec(
             console.print(f"\nInstalling airflow from GitHub PR #{use_airflow_version}: {resolved_version}\n")
         else:
             # Handle owner/repo:branch format
+            if repo_match is None:
+                console.print(
+                    f"[red]USE_AIRFLOW_VERSION '{use_airflow_version}' did not match expected "
+                    "owner/repo:branch pattern"
+                )
+                sys.exit(1)
             owner, repo, branch = repo_match.groups()
             resolved_version = use_airflow_version
             console.print(f"\nInstalling airflow from GitHub: {use_airflow_version}\n")