From e259efa403b198826b1fe4d4b03daa893227f84a Mon Sep 17 00:00:00 2001
From: John Chilton <jmchilton@gmail.com>
Date: Wed, 24 Apr 2024 12:21:26 -0400
Subject: [PATCH] WIP: file source templates

---
 client/src/api/schema/schema.ts               | 252 ++++++++++-
 .../ConfigTemplates/ManageIndexHeader.vue     |  35 ++
 .../src/components/ConfigTemplates/fields.ts  |  25 ++
 .../src/components/ConfigTemplates/icons.ts   |   5 +
 .../FileSources/Instances/ManageIndex.vue     |  68 +++
 .../components/FileSources/ManageIndex.vue    |   0
 .../ObjectStore/Instances/ManageIndex.vue     |  65 +--
 .../components/ObjectStore/Instances/types.ts |   4 +-
 .../components/ObjectStore/Instances/util.ts  |   6 +-
 .../src/components/User/UserPreferences.vue   |  28 +-
 client/src/entry/analysis/router.js           |  12 +-
 client/src/stores/configTemplatesUtil.ts      |  66 +++
 client/src/stores/fileSourceInstancesStore.ts |  55 +++
 client/src/stores/fileSourceTemplatesStore.ts |  70 +++
 .../src/stores/objectStoreTemplatesStore.ts   |  49 +-
 lib/galaxy/app.py                             |  39 +-
 lib/galaxy/app_unittest_utils/galaxy_mock.py  |  22 +
 lib/galaxy/config/schemas/config_schema.yml   |  15 +
 lib/galaxy/files/__init__.py                  | 136 +++++-
 lib/galaxy/files/plugins.py                   |   5 +
 lib/galaxy/files/sources/__init__.py          |  11 +-
 lib/galaxy/files/templates/__init__.py        |  17 +
 lib/galaxy/files/templates/manager.py         |  86 ++++
 lib/galaxy/files/templates/models.py          | 125 ++++++
 lib/galaxy/files/unittest_utils/__init__.py   |   7 +-
 lib/galaxy/managers/file_source_instances.py  | 420 ++++++++++++++++++
 lib/galaxy/managers/object_store_instances.py |  36 +-
 lib/galaxy/model/__init__.py                  |  86 +++-
 ...a3c93d66b_add_user_defined_file_sources.py |  57 +++
 lib/galaxy/objectstore/templates/manager.py   |  92 +---
 lib/galaxy/objectstore/templates/models.py    |  59 +--
 lib/galaxy/util/config_templates.py           | 167 +++++++
 lib/galaxy/webapps/galaxy/api/file_sources.py |  95 ++++
 lib/galaxy/webapps/galaxy/buildapp.py         |   5 +
 packages/util/setup.cfg                       |   2 +
 .../app/managers/test_user_file_sources.py    | 109 +++++
 test/unit/files/_util.py                      |   3 +-
 test/unit/files/test_posix.py                 |  29 +-
 test/unit/files/test_template_manager.py      |  58 +++
 test/unit/files/test_template_models.py       |  91 ++++
 40 files changed, 2206 insertions(+), 306 deletions(-)
 create mode 100644 client/src/components/ConfigTemplates/ManageIndexHeader.vue
 create mode 100644 client/src/components/ConfigTemplates/fields.ts
 create mode 100644 client/src/components/ConfigTemplates/icons.ts
 create mode 100644 client/src/components/FileSources/Instances/ManageIndex.vue
 create mode 100644 client/src/components/FileSources/ManageIndex.vue
 create mode 100644 client/src/stores/configTemplatesUtil.ts
 create mode 100644 client/src/stores/fileSourceInstancesStore.ts
 create mode 100644 client/src/stores/fileSourceTemplatesStore.ts
 create mode 100644 lib/galaxy/files/templates/__init__.py
 create mode 100644 lib/galaxy/files/templates/manager.py
 create mode 100644 lib/galaxy/files/templates/models.py
 create mode 100644 lib/galaxy/managers/file_source_instances.py
 create mode 100644 lib/galaxy/model/migrations/alembic/versions_gxy/c14a3c93d66b_add_user_defined_file_sources.py
 create mode 100644 lib/galaxy/util/config_templates.py
 create mode 100644 lib/galaxy/webapps/galaxy/api/file_sources.py
 create mode 100644 test/unit/app/managers/test_user_file_sources.py
 create mode 100644 test/unit/files/test_template_manager.py
 create mode 100644 test/unit/files/test_template_models.py

diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts
index 67b97e8f6d7e..503f64a5e93f 100644
--- a/client/src/api/schema/schema.ts
+++ b/client/src/api/schema/schema.ts
@@ -318,6 +318,22 @@ export interface paths {
         /** Download */
         get: operations["download_api_drs_download__object_id__get"];
     };
+    "/api/file_source_instances": {
+        /** Get a list of persisted file source instances defined by the requesting user. */
+        get: operations["file_sources__instances_index"];
+        /** Create a user-bound object store. */
+        post: operations["file_sources__create_instance"];
+    };
+    "/api/file_source_instances/{user_file_source_id}": {
+        /** Get a list of persisted file source instances defined by the requesting user. */
+        get: operations["file_sources__instances_get"];
+        /** Update or upgrade user file source instance. */
+        put: operations["file_sources__instances_update"];
+    };
+    "/api/file_source_templates": {
+        /** Get a list of file source templates available to build user defined file sources from */
+        get: operations["file_sources__templates_index"];
+    };
     "/api/folders/{folder_id}/contents": {
         /**
          * Returns a list of a folder's contents (files and sub-folders) with additional metadata about the folder.
@@ -5230,6 +5246,36 @@ export interface components {
              */
             update_time: string;
         };
+        /** FileSourceTemplateSummaries */
+        FileSourceTemplateSummaries: components["schemas"]["FileSourceTemplateSummary"][];
+        /** FileSourceTemplateSummary */
+        FileSourceTemplateSummary: {
+            /** Description */
+            description: string | null;
+            /**
+             * Hidden
+             * @default false
+             */
+            hidden?: boolean;
+            /** Id */
+            id: string;
+            /** Name */
+            name: string | null;
+            /** Secrets */
+            secrets?: components["schemas"]["TemplateSecret"][] | null;
+            /**
+             * Type
+             * @enum {string}
+             */
+            type: "posix" | "s3fs";
+            /** Variables */
+            variables?: components["schemas"]["TemplateVariable"][] | null;
+            /**
+             * Version
+             * @default 0
+             */
+            version?: number;
+        };
         /** FilesSourcePlugin */
         FilesSourcePlugin: {
             /**
@@ -9909,13 +9955,6 @@ export interface components {
              */
             up_to_date: boolean;
         };
-        /** ObjectStoreTemplateSecret */
-        ObjectStoreTemplateSecret: {
-            /** Help */
-            help: string | null;
-            /** Name */
-            name: string;
-        };
         /** ObjectStoreTemplateSummaries */
         ObjectStoreTemplateSummaries: components["schemas"]["ObjectStoreTemplateSummary"][];
         /** ObjectStoreTemplateSummary */
@@ -9934,32 +9973,20 @@ export interface components {
             /** Name */
             name: string | null;
             /** Secrets */
-            secrets?: components["schemas"]["ObjectStoreTemplateSecret"][] | null;
+            secrets?: components["schemas"]["TemplateSecret"][] | null;
             /**
              * Type
              * @enum {string}
              */
             type: "s3" | "azure_blob" | "disk" | "generic_s3";
             /** Variables */
-            variables?: components["schemas"]["ObjectStoreTemplateVariable"][] | null;
+            variables?: components["schemas"]["TemplateVariable"][] | null;
             /**
              * Version
              * @default 0
              */
             version?: number;
         };
-        /** ObjectStoreTemplateVariable */
-        ObjectStoreTemplateVariable: {
-            /** Help */
-            help: string | null;
-            /** Name */
-            name: string;
-            /**
-             * Type
-             * @enum {string}
-             */
-            type: "string" | "boolean" | "integer";
-        };
         /** OutputReferenceByLabel */
         OutputReferenceByLabel: {
             /**
@@ -11836,6 +11863,25 @@ export interface components {
          * @enum {string}
          */
         TaskState: "PENDING" | "STARTED" | "RETRY" | "FAILURE" | "SUCCESS";
+        /** TemplateSecret */
+        TemplateSecret: {
+            /** Help */
+            help: string | null;
+            /** Name */
+            name: string;
+        };
+        /** TemplateVariable */
+        TemplateVariable: {
+            /** Help */
+            help: string | null;
+            /** Name */
+            name: string;
+            /**
+             * Type
+             * @enum {string}
+             */
+            type: "string" | "boolean" | "integer";
+        };
         /** ToolDataDetails */
         ToolDataDetails: {
             /**
@@ -12566,6 +12612,34 @@ export interface components {
              */
             id: string;
         };
+        /** UserFileSourceModel */
+        UserFileSourceModel: {
+            /** Description */
+            description: string | null;
+            /** Id */
+            id: string | number;
+            /** Name */
+            name: string;
+            /** Secrets */
+            secrets: string[];
+            /** Template Id */
+            template_id: string;
+            /** Template Version */
+            template_version: number;
+            /**
+             * Type
+             * @enum {string}
+             */
+            type: "posix" | "s3fs";
+            /** Uri Root */
+            uri_root: string;
+            /** Uuid */
+            uuid: string;
+            /** Variables */
+            variables: {
+                [key: string]: (string | boolean | number) | undefined;
+            } | null;
+        };
         /**
          * UserModel
          * @description User in a transaction context.
@@ -14755,6 +14829,142 @@ export interface operations {
             };
         };
     };
+    file_sources__instances_index: {
+        /** Get a list of persisted file source instances defined by the requesting user. */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["UserFileSourceModel"][];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    file_sources__create_instance: {
+        /** Create a user-bound object store. */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+        };
+        requestBody: {
+            content: {
+                "application/json": components["schemas"]["CreateInstancePayload"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["UserFileSourceModel"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    file_sources__instances_get: {
+        /** Get a list of persisted file source instances defined by the requesting user. */
+        parameters: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+            /** @description The index for a persisted UserFileSourceStore object. */
+            path: {
+                user_file_source_id: string;
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["UserFileSourceModel"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    file_sources__instances_update: {
+        /** Update or upgrade user file source instance. */
+        parameters: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+            /** @description The index for a persisted UserFileSourceStore object. */
+            path: {
+                user_file_source_id: string;
+            };
+        };
+        requestBody: {
+            content: {
+                "application/json":
+                    | components["schemas"]["UpdateInstanceSecretPayload"]
+                    | components["schemas"]["UpgradeInstancePayload"]
+                    | components["schemas"]["UpdateInstancePayload"];
+            };
+        };
+        responses: {
+            /** @description Successful Response */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["UserFileSourceModel"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    file_sources__templates_index: {
+        /** Get a list of file source templates available to build user defined file sources from */
+        parameters?: {
+            /** @description The user ID that will be used to effectively make this API call. Only admins and designated users can make API calls on behalf of other users. */
+            header?: {
+                "run-as"?: string | null;
+            };
+        };
+        responses: {
+            /** @description A list of the configured file source templates. */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["FileSourceTemplateSummaries"];
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     index_api_folders__folder_id__contents_get: {
         /**
          * Returns a list of a folder's contents (files and sub-folders) with additional metadata about the folder.
diff --git a/client/src/components/ConfigTemplates/ManageIndexHeader.vue b/client/src/components/ConfigTemplates/ManageIndexHeader.vue
new file mode 100644
index 000000000000..2f03901f579a
--- /dev/null
+++ b/client/src/components/ConfigTemplates/ManageIndexHeader.vue
@@ -0,0 +1,35 @@
+<script setup lang="ts">
+import "./icons";
+
+import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome";
+import { BButton, BCol, BRow } from "bootstrap-vue";
+import { useRouter } from "vue-router/composables";
+
+import _l from "@/utils/localization";
+
+interface Props {
+    message: String | null | undefined;
+    createButtonId: string;
+    createRoute: string;
+}
+
+defineProps<Props>();
+
+const router = useRouter();
+</script>
+
+<template>
+    <div>
+        <p>
+            {{ message || "" }}
+        </p>
+        <BRow class="mb-3">
+            <BCol>
+                <BButton :id="createButtonId" class="m-1 float-right" @click="router.push(createRoute)">
+                    <FontAwesomeIcon icon="plus" />
+                    {{ _l("Create") }}
+                </BButton>
+            </BCol>
+        </BRow>
+    </div>
+</template>
diff --git a/client/src/components/ConfigTemplates/fields.ts b/client/src/components/ConfigTemplates/fields.ts
new file mode 100644
index 000000000000..33a00f7e69f4
--- /dev/null
+++ b/client/src/components/ConfigTemplates/fields.ts
@@ -0,0 +1,25 @@
+import _l from "@/utils/localization";
+
+export const NAME_FIELD = {
+    key: "name",
+    label: _l("Name"),
+    sortable: true,
+};
+
+export const DESCRIPTION_FIELD = {
+    key: "description",
+    label: _l("Description"),
+    sortable: true,
+};
+
+export const TYPE_FIELD = {
+    key: "type",
+    label: _l("Type"),
+    sortable: true,
+};
+
+export const TEMPLATE_FIELD = {
+    key: "template",
+    label: _l("From Template"),
+    sortable: true,
+};
diff --git a/client/src/components/ConfigTemplates/icons.ts b/client/src/components/ConfigTemplates/icons.ts
new file mode 100644
index 000000000000..0c95a604c25c
--- /dev/null
+++ b/client/src/components/ConfigTemplates/icons.ts
@@ -0,0 +1,5 @@
+import { library } from "@fortawesome/fontawesome-svg-core";
+import { faCaretDown, faPlus } from "@fortawesome/free-solid-svg-icons";
+
+library.add(faCaretDown);
+library.add(faPlus);
diff --git a/client/src/components/FileSources/Instances/ManageIndex.vue b/client/src/components/FileSources/Instances/ManageIndex.vue
new file mode 100644
index 000000000000..6331f5f8a236
--- /dev/null
+++ b/client/src/components/FileSources/Instances/ManageIndex.vue
@@ -0,0 +1,68 @@
+<script setup lang="ts">
+import "@/components/ConfigTemplates/icons";
+
+import { BTable } from "bootstrap-vue";
+import { computed } from "vue";
+
+import { DESCRIPTION_FIELD, NAME_FIELD, TEMPLATE_FIELD, TYPE_FIELD } from "@/components/ConfigTemplates/fields";
+import { useFileSourceInstancesStore } from "@/stores/fileSourceInstancesStore";
+
+import ManageIndexHeader from "@/components/ConfigTemplates/ManageIndexHeader.vue";
+import LoadingSpan from "@/components/LoadingSpan.vue";
+import InstanceDropdown from "@/components/ObjectStore/Instances/InstanceDropdown.vue";
+import ObjectStoreTypeSpan from "@/components/ObjectStore/ObjectStoreTypeSpan.vue";
+import TemplateSummarySpan from "@/components/ObjectStore/Templates/TemplateSummarySpan.vue";
+
+const fileSourceInstancesStore = useFileSourceInstancesStore();
+
+interface Props {
+    message: String | undefined | null;
+}
+
+defineProps<Props>();
+
+const fields = [NAME_FIELD, DESCRIPTION_FIELD, TYPE_FIELD, TEMPLATE_FIELD];
+
+const items = computed(() => fileSourceInstancesStore.getInstances);
+const loading = computed(() => fileSourceInstancesStore.loading);
+fileSourceInstancesStore.fetchInstances();
+</script>
+
+<template>
+    <div>
+        <ManageIndexHeader
+            :message="message"
+            create-button-id="object-store-create"
+            create-route="/object_store_instances/create">
+        </ManageIndexHeader>
+        <BTable
+            no-sort-reset
+            :fields="fields"
+            :items="items"
+            :hover="true"
+            :striped="true"
+            :caption-top="true"
+            :fixed="true"
+            :show-empty="true">
+            <template v-slot:empty>
+                <LoadingSpan v-if="loading" message="Loading your user's file source instances" />
+                <b-alert v-else id="no-file-source-instances" variant="info" show>
+                    <div>
+                        No file source instances found for your users, click the create button to configure a new one.
+                    </div>
+                </b-alert>
+            </template>
+            <template v-slot:cell(name)="row">
+                <InstanceDropdown :object-store="row.item" />
+            </template>
+            <template v-slot:cell(type)="row">
+                <ObjectStoreTypeSpan :type="row.item.type" />
+            </template>
+            <template v-slot:cell(template)="row">
+                <TemplateSummarySpan
+                    :template-version="row.item.template_version ?? 0"
+                    :template-id="row.item.template_id" />
+            </template>
+        </BTable>
+    </div>
+</template>
diff --git a/client/src/components/FileSources/ManageIndex.vue b/client/src/components/FileSources/ManageIndex.vue
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/client/src/components/ObjectStore/Instances/ManageIndex.vue b/client/src/components/ObjectStore/Instances/ManageIndex.vue
index 9695f492dad2..1f87d85ebc09 100644
--- a/client/src/components/ObjectStore/Instances/ManageIndex.vue
+++ b/client/src/components/ObjectStore/Instances/ManageIndex.vue
@@ -1,20 +1,20 @@
 <script setup lang="ts">
-import "./icons";
+import "@/components/ConfigTemplates/icons";
 
-import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome";
+import { BTable } from "bootstrap-vue";
 import { computed } from "vue";
-import { useRouter } from "vue-router/composables";
 
+import { DESCRIPTION_FIELD, NAME_FIELD, TEMPLATE_FIELD, TYPE_FIELD } from "@/components/ConfigTemplates/fields";
 import { useObjectStoreInstancesStore } from "@/stores/objectStoreInstancesStore";
 import _l from "@/utils/localization";
 
 import InstanceDropdown from "./InstanceDropdown.vue";
+import ManageIndexHeader from "@/components/ConfigTemplates/ManageIndexHeader.vue";
 import LoadingSpan from "@/components/LoadingSpan.vue";
 import ObjectStoreBadges from "@/components/ObjectStore/ObjectStoreBadges.vue";
 import ObjectStoreTypeSpan from "@/components/ObjectStore/ObjectStoreTypeSpan.vue";
 import TemplateSummarySpan from "@/components/ObjectStore/Templates/TemplateSummarySpan.vue";
 
-const router = useRouter();
 const objectStoreInstancesStore = useObjectStoreInstancesStore();
 
 interface Props {
@@ -23,33 +23,13 @@ interface Props {
 
 defineProps<Props>();
 
-const fields = [
-    {
-        key: "name",
-        label: _l("Name"),
-        sortable: true,
-    },
-    {
-        key: "description",
-        label: _l("Description"),
-        sortable: true,
-    },
-    {
-        key: "type",
-        label: _l("Type"),
-        sortable: true,
-    },
-    {
-        key: "template",
-        label: _l("From Template"),
-        sortable: true,
-    },
-    {
-        key: "badges",
-        label: _l(" "),
-        sortable: false,
-    },
-];
+const BADGE_FIELD = {
+    key: "badges",
+    label: _l(" "),
+    sortable: false,
+};
+
+const fields = [NAME_FIELD, DESCRIPTION_FIELD, TYPE_FIELD, TEMPLATE_FIELD, BADGE_FIELD];
 
 const items = computed(() => objectStoreInstancesStore.getInstances);
 const loading = computed(() => objectStoreInstancesStore.loading);
@@ -58,21 +38,12 @@ objectStoreInstancesStore.fetchInstances();
 
 <template>
     <div>
-        <p>
-            {{ message || "" }}
-        </p>
-        <b-row class="mb-3">
-            <b-col>
-                <b-button
-                    id="object-store-create"
-                    class="m-1 float-right"
-                    @click="router.push('/object_store_instances/create')">
-                    <FontAwesomeIcon icon="plus" />
-                    {{ _l("Create") }}
-                </b-button>
-            </b-col>
-        </b-row>
-        <b-table
+        <ManageIndexHeader
+            :message="message"
+            create-button-id="object-store-create"
+            create-route="/object_store_instances/create">
+        </ManageIndexHeader>
+        <BTable
             no-sort-reset
             :fields="fields"
             :items="items"
@@ -101,6 +72,6 @@ objectStoreInstancesStore.fetchInstances();
                     :template-version="row.item.template_version ?? 0"
                     :template-id="row.item.template_id" />
             </template>
-        </b-table>
+        </BTable>
     </div>
 </template>
diff --git a/client/src/components/ObjectStore/Instances/types.ts b/client/src/components/ObjectStore/Instances/types.ts
index bd956468c7b8..887b4c4468c9 100644
--- a/client/src/components/ObjectStore/Instances/types.ts
+++ b/client/src/components/ObjectStore/Instances/types.ts
@@ -2,8 +2,8 @@ import type { components } from "@/api/schema/schema";
 
 export type UserConcreteObjectStore = components["schemas"]["UserConcreteObjectStoreModel"];
 export type CreateInstancePayload = components["schemas"]["CreateInstancePayload"];
-export type ObjectStoreTemplateVariable = components["schemas"]["ObjectStoreTemplateVariable"];
-export type ObjectStoreTemplateSecret = components["schemas"]["ObjectStoreTemplateSecret"];
+export type TemplateVariable = components["schemas"]["TemplateVariable"];
+export type TemplateSecret = components["schemas"]["TemplateSecret"];
 export type VariableValueType = (string | boolean | number) | undefined;
 export type VariableData = { [key: string]: VariableValueType };
 export type SecretData = { [key: string]: string };
diff --git a/client/src/components/ObjectStore/Instances/util.ts b/client/src/components/ObjectStore/Instances/util.ts
index 2ba02b8a75f2..5f2b8fcaba23 100644
--- a/client/src/components/ObjectStore/Instances/util.ts
+++ b/client/src/components/ObjectStore/Instances/util.ts
@@ -1,6 +1,6 @@
 import { markup } from "@/components/ObjectStore/configurationMarkdown";
 
-import type { ObjectStoreTemplateSecret, ObjectStoreTemplateVariable, VariableValueType } from "./types";
+import type { TemplateSecret, TemplateVariable, VariableValueType } from "./types";
 
 export function metadataFormEntryName() {
     return {
@@ -22,7 +22,7 @@ export function metadataFormEntryDescription() {
     };
 }
 
-export function templateVariableFormEntry(variable: ObjectStoreTemplateVariable, variableValue: VariableValueType) {
+export function templateVariableFormEntry(variable: TemplateVariable, variableValue: VariableValueType) {
     return {
         name: variable.name,
         type: "text",
@@ -31,7 +31,7 @@ export function templateVariableFormEntry(variable: ObjectStoreTemplateVariable,
     };
 }
 
-export function templateSecretFormEntry(secret: ObjectStoreTemplateSecret) {
+export function templateSecretFormEntry(secret: TemplateSecret) {
     return {
         name: secret.name,
         type: "password",
diff --git a/client/src/components/User/UserPreferences.vue b/client/src/components/User/UserPreferences.vue
index 6807bda7f47a..47197e0ad92e 100644
--- a/client/src/components/User/UserPreferences.vue
+++ b/client/src/components/User/UserPreferences.vue
@@ -93,13 +93,21 @@
             :user-id="userId">
         </UserPreferredObjectStore>
         <UserPreferencesElement
-            v-if="hasTemplates"
+            v-if="hasObjectStoreTemplates"
             id="manage-object-stores"
             class="manage-object-stores"
             icon="fa-hdd"
             title="Manage Your Storage Locations"
             description="Add, remove, or update your personally configured storage locations."
             to="/object_store_instances/index" />
+        <UserPreferencesElement
+            v-if="hasFileSourceTemplates"
+            id="manage-file-sources"
+            class="manage-file-sources"
+            icon="fa-file"
+            title="Manage Your Remote File Sources"
+            description="Add, remove, or update your personally configured location to find files from and write files to."
+            to="/file_source_instances/index" />
         <UserDeletion
             v-if="isConfigLoaded && !config.single_user && config.enable_account_interface"
             :email="email"
@@ -144,6 +152,7 @@ import { withPrefix } from "utils/redirect";
 import Vue from "vue";
 
 import { useConfig } from "@/composables/config";
+import { useFileSourceTemplatesStore } from "@/stores/fileSourceTemplatesStore";
 import { useObjectStoreTemplatesStore } from "@/stores/objectStoreTemplatesStore";
 import { useUserStore } from "@/stores/userStore";
 
@@ -192,7 +201,12 @@ export default {
     },
     computed: {
         ...mapState(useUserStore, ["currentUser"]),
-        ...mapState(useObjectStoreTemplatesStore, ["hasTemplates"]),
+        ...mapState(useObjectStoreTemplatesStore, {
+            hasObjectStoreTemplates: "hasTemplates",
+        }),
+        ...mapState(useFileSourceTemplatesStore, {
+            hasFileSourceTemplates: "hasTemplates",
+        }),
         activePreferences() {
             const userPreferencesEntries = Object.entries(getUserPreferencesModel());
             // Object.entries returns an array of arrays, where the first element
@@ -232,10 +246,16 @@ export default {
             this.diskUsage = response.data.nice_total_disk_usage;
             this.diskQuota = response.data.quota;
         });
-        this.ensureTemplates();
+        this.ensureObjectStoreTemplates();
+        this.ensureFileSourceTemplates();
     },
     methods: {
-        ...mapActions(useObjectStoreTemplatesStore, ["ensureTemplates"]),
+        ...mapActions(useObjectStoreTemplatesStore, {
+            ensureObjectStoreTemplates: "ensureTemplates",
+        }),
+        ...mapActions(useFileSourceTemplatesStore, {
+            ensureFileSourceTemplates: "ensureTemplates",
+        }),
         toggleNotifications() {
             if (window.Notification) {
                 Notification.requestPermission().then(function (permission) {
diff --git a/client/src/entry/analysis/router.js b/client/src/entry/analysis/router.js
index 320a6ec8fbb7..964003b8599b 100644
--- a/client/src/entry/analysis/router.js
+++ b/client/src/entry/analysis/router.js
@@ -63,12 +63,13 @@ import { parseBool } from "@/utils/utils";
 import { patchRouterPush } from "./router-push";
 
 import AboutGalaxy from "@/components/AboutGalaxy.vue";
+import ManageFileSourceIndex from "@/components/FileSources/Instances/ManageIndex.vue";
 import GridVisualization from "@/components/Grid/GridVisualization.vue";
 import HistoryArchiveWizard from "@/components/History/Archiving/HistoryArchiveWizard.vue";
 import HistoryDatasetPermissions from "@/components/History/HistoryDatasetPermissions.vue";
 import NotificationsList from "@/components/Notifications/NotificationsList.vue";
 import EditInstance from "@/components/ObjectStore/Instances/EditInstance.vue";
-import ManageIndex from "@/components/ObjectStore/Instances/ManageIndex.vue";
+import ManageObjectStoreIndex from "@/components/ObjectStore/Instances/ManageIndex.vue";
 import UpgradeInstance from "@/components/ObjectStore/Instances/UpgradeInstance.vue";
 import CreateUserObjectStore from "@/components/ObjectStore/Templates/CreateUserObjectStore.vue";
 import Sharing from "@/components/Sharing/SharingPage.vue";
@@ -355,7 +356,7 @@ export function getRouter(Galaxy) {
                     },
                     {
                         path: "object_store_instances/index",
-                        component: ManageIndex,
+                        component: ManageObjectStoreIndex,
                         props: (route) => {
                             return { message: route.query["message"] };
                         },
@@ -375,6 +376,13 @@ export function getRouter(Galaxy) {
                         component: CreateInstance,
                         props: true,
                     },
+                    {
+                        path: "file_source_instances/index",
+                        component: ManageFileSourceIndex,
+                        props: (route) => {
+                            return { message: route.query["message"] };
+                        },
+                    },
                     {
                         path: "pages/create",
                         component: FormGeneric,
diff --git a/client/src/stores/configTemplatesUtil.ts b/client/src/stores/configTemplatesUtil.ts
new file mode 100644
index 000000000000..01aaf7985867
--- /dev/null
+++ b/client/src/stores/configTemplatesUtil.ts
@@ -0,0 +1,66 @@
+import type { components } from "@/api/schema/schema";
+
+export interface TemplateSummary {
+    description: string | null;
+    hidden?: boolean;
+    id: string;
+    name: string | null;
+    secrets?: components["schemas"]["TemplateSecret"][] | null;
+    variables?: components["schemas"]["TemplateVariable"][] | null;
+    version?: number;
+}
+
+export function findTemplate<T extends TemplateSummary>(
+    templates: T[],
+    templateId: string,
+    templateVersion: number
+): T | null {
+    for (const template of templates) {
+        if (template.id == templateId && template.version == templateVersion) {
+            return template;
+        }
+    }
+    return null;
+}
+
+export function getLatestVersionMap<T extends TemplateSummary>(templates: T[]): { [key: string]: number } {
+    const latestVersions: { [key: string]: number } = {};
+    templates.forEach((i: T) => {
+        const templateId = i.id;
+        const templateVersion = i.version || 0;
+        if ((latestVersions[templateId] ?? -1) < templateVersion) {
+            latestVersions[templateId] = templateVersion;
+        }
+    });
+    return latestVersions;
+}
+
+export function canUpgrade<T extends TemplateSummary>(
+    templates: T[],
+    templateId: string,
+    templateVersion: number
+): boolean {
+    let can = false;
+    templates.forEach((i: T) => {
+        if (i.id == templateId && i.version && i.version > templateVersion) {
+            can = true;
+        }
+    });
+    return can;
+}
+
+export function getLatestVersion<T extends TemplateSummary>(templates: T[], id: string): T | null {
+    let latestVersion = -1;
+    let latestTemplate = null as T | null;
+    templates.forEach((i: T) => {
+        const templateId = i.id;
+        if (templateId == id) {
+            const templateVersion = i.version || 0;
+            if (templateVersion > latestVersion) {
+                latestTemplate = i;
+                latestVersion = templateVersion;
+            }
+        }
+    });
+    return latestTemplate;
+}
diff --git a/client/src/stores/fileSourceInstancesStore.ts b/client/src/stores/fileSourceInstancesStore.ts
new file mode 100644
index 000000000000..4f9cdb156e31
--- /dev/null
+++ b/client/src/stores/fileSourceInstancesStore.ts
@@ -0,0 +1,55 @@
+import { defineStore } from "pinia";
+
+import { fetcher } from "@/api/schema/fetcher";
+import type { components } from "@/api/schema/schema";
+import { errorMessageAsString } from "@/utils/simple-error";
+
+const getFileSourceInstances = fetcher.path("/api/file_source_instances").method("get").create();
+
+type UserFileSourceModel = components["schemas"]["UserFileSourceModel"];
+
+function _asNumber(x: number | string): number {
+    return +x;
+}
+
+export const useFileSourceInstancesStore = defineStore("fileSourceInstances", {
+    state: () => ({
+        instances: [] as UserFileSourceModel[],
+        fetched: false,
+        error: null as string | null,
+    }),
+    getters: {
+        getInstances: (state) => {
+            return state.instances;
+        },
+        loading: (state) => {
+            return !state.fetched;
+        },
+        getInstance: (state) => {
+            return (id: number | string) => state.instances.find((i) => i.id == _asNumber(id));
+        },
+    },
+    actions: {
+        async handleInit(instances: UserFileSourceModel[]) {
+            this.instances = instances;
+            this.fetched = true;
+            this.error = null;
+        },
+        async handleError(err: unknown) {
+            this.error = errorMessageAsString(err);
+        },
+        async fetchInstances() {
+            try {
+                const { data: instances } = await getFileSourceInstances({});
+                this.handleInit(instances);
+            } catch (err) {
+                this.handleError(err);
+            }
+        },
+        async ensureTemplates() {
+            if (!this.fetched || this.error != null) {
+                await this.fetchInstances();
+            }
+        },
+    },
+});
diff --git a/client/src/stores/fileSourceTemplatesStore.ts b/client/src/stores/fileSourceTemplatesStore.ts
new file mode 100644
index 000000000000..1efd7e67bfe0
--- /dev/null
+++ b/client/src/stores/fileSourceTemplatesStore.ts
@@ -0,0 +1,70 @@
+import { defineStore } from "pinia";
+
+import { fetcher } from "@/api/schema/fetcher";
+import type { components } from "@/api/schema/schema";
+import { errorMessageAsString } from "@/utils/simple-error";
+
+import { canUpgrade, findTemplate, getLatestVersion, getLatestVersionMap } from "./configTemplatesUtil";
+
+const getFileSourceTemplates = fetcher.path("/api/file_source_templates").method("get").create();
+
+type FileSourceTemplateSummary = components["schemas"]["FileSourceTemplateSummary"];
+type FileSourceTemplateSummaries = FileSourceTemplateSummary[];
+
+export const useFileSourceTemplatesStore = defineStore("fileSourceTemplatesStore", {
+    state: () => ({
+        templates: [] as FileSourceTemplateSummaries,
+        fetched: false,
+        error: null as string | null,
+    }),
+    getters: {
+        latestTemplates: (state) => {
+            // only expose latest instance by template_version for each template_id
+            const latestVersions = getLatestVersionMap(state.templates);
+            return state.templates.filter((i: FileSourceTemplateSummary) => latestVersions[i.id] == (i.version || 0));
+        },
+        canUpgrade: (state) => {
+            return (templateId: string, templateVersion: number) =>
+                canUpgrade(state.templates, templateId, templateVersion);
+        },
+        getTemplates: (state) => {
+            return state.templates;
+        },
+        getTemplate: (state) => {
+            return (templateId: string, templateVersion: number) =>
+                findTemplate(state.templates, templateId, templateVersion);
+        },
+        getLatestTemplate: (state) => {
+            return (templateId: string) => getLatestVersion(state.templates, templateId);
+        },
+        hasTemplates: (state) => {
+            return state.templates.length > 0;
+        },
+        loading: (state) => {
+            return !state.fetched;
+        },
+    },
+    actions: {
+        async handleInit(templates: FileSourceTemplateSummaries) {
+            this.templates = templates;
+            this.fetched = true;
+        },
+        async handleError(err: unknown) {
+            this.fetched = true;
+            this.error = errorMessageAsString(err);
+        },
+        async fetchTemplates() {
+            try {
+                const { data: templates } = await getFileSourceTemplates({});
+                this.handleInit(templates);
+            } catch (err) {
+                this.handleError(err);
+            }
+        },
+        async ensureTemplates() {
+            if (!this.fetched || this.error != null) {
+                await this.fetchTemplates();
+            }
+        },
+    },
+});
diff --git a/client/src/stores/objectStoreTemplatesStore.ts b/client/src/stores/objectStoreTemplatesStore.ts
index c7c340a50f08..4b9c852d494f 100644
--- a/client/src/stores/objectStoreTemplatesStore.ts
+++ b/client/src/stores/objectStoreTemplatesStore.ts
@@ -4,58 +4,13 @@ import { fetcher } from "@/api/schema/fetcher";
 import type { components } from "@/api/schema/schema";
 import { errorMessageAsString } from "@/utils/simple-error";
 
+import { canUpgrade, findTemplate, getLatestVersion, getLatestVersionMap } from "./configTemplatesUtil";
+
 const getObjectStoreTemplates = fetcher.path("/api/object_store_templates").method("get").create();
 
 type ObjectStoreTemplateSummary = components["schemas"]["ObjectStoreTemplateSummary"];
 type ObjectStoreTemplateSummaries = ObjectStoreTemplateSummary[];
 
-function findTemplate(templates: ObjectStoreTemplateSummaries, templateId: string, templateVersion: number) {
-    for (const template of templates) {
-        if (template.id == templateId && template.version == templateVersion) {
-            return template;
-        }
-    }
-    return null;
-}
-
-function getLatestVersionMap(templates: ObjectStoreTemplateSummaries): { [key: string]: number } {
-    const latestVersions: { [key: string]: number } = {};
-    templates.forEach((i: ObjectStoreTemplateSummary) => {
-        const templateId = i.id;
-        const templateVersion = i.version || 0;
-        if ((latestVersions[templateId] ?? -1) < templateVersion) {
-            latestVersions[templateId] = templateVersion;
-        }
-    });
-    return latestVersions;
-}
-
-function canUpgrade(templates: ObjectStoreTemplateSummaries, templateId: string, templateVersion: number): boolean {
-    let can = false;
-    templates.forEach((i: ObjectStoreTemplateSummary) => {
-        if (i.id == templateId && i.version && i.version > templateVersion) {
-            can = true;
-        }
-    });
-    return can;
-}
-
-function getLatestVersion(templates: ObjectStoreTemplateSummaries, id: string): ObjectStoreTemplateSummary | null {
-    let latestVersion = -1;
-    let latestTemplate = null as ObjectStoreTemplateSummary | null;
-    templates.forEach((i: ObjectStoreTemplateSummary) => {
-        const templateId = i.id;
-        if (templateId == id) {
-            const templateVersion = i.version || 0;
-            if (templateVersion > latestVersion) {
-                latestTemplate = i;
-                latestVersion = templateVersion;
-            }
-        }
-    });
-    return latestTemplate;
-}
-
 export const useObjectStoreTemplatesStore = defineStore("objectStoreTemplatesStore", {
     state: () => ({
         templates: [] as ObjectStoreTemplateSummaries,
diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py
index 682b85fc751b..782a6ef868f3 100644
--- a/lib/galaxy/app.py
+++ b/lib/galaxy/app.py
@@ -35,13 +35,27 @@
 )
 from galaxy.config_watchers import ConfigWatchers
 from galaxy.datatypes.registry import Registry
-from galaxy.files import ConfiguredFileSources
+from galaxy.files import (
+    ConfiguredFileSources,
+    ConfiguredFileSourcesConf,
+    UserDefinedFileSources,
+)
+from galaxy.files.plugins import (
+    FileSourcePluginLoader,
+    FileSourcePluginsConfig,
+)
+from galaxy.files.templates import ConfiguredFileSourceTemplates
 from galaxy.job_metrics import JobMetrics
 from galaxy.jobs.manager import JobManager
 from galaxy.managers.api_keys import ApiKeyManager
 from galaxy.managers.citations import CitationsManager
 from galaxy.managers.collections import DatasetCollectionManager
 from galaxy.managers.dbkeys import GenomeBuilds
+from galaxy.managers.file_source_instances import (
+    FileSourceInstancesManager,
+    UserDefinedFileSourcesConfig,
+    UserDefinedFileSourcesImpl,
+)
 from galaxy.managers.folders import FolderManager
 from galaxy.managers.hdas import HDAManager
 from galaxy.managers.histories import HistoryManager
@@ -585,9 +599,28 @@ def __init__(self, configure_logging=True, use_converters=True, use_display_appl
         )
 
         # ConfiguredFileSources
-        self.file_sources = self._register_singleton(
-            ConfiguredFileSources, ConfiguredFileSources.from_app_config(self.config)
+        vault_configured = is_vault_configured(self.vault)
+        file_sources_config: FileSourcePluginsConfig = FileSourcePluginsConfig.from_app_config(self.config)
+        self._register_singleton(FileSourcePluginsConfig, file_sources_config)
+        file_source_plugin_loader = FileSourcePluginLoader()
+        self._register_singleton(FileSourcePluginLoader, file_source_plugin_loader)
+        self._register_singleton(
+            UserDefinedFileSourcesConfig, UserDefinedFileSourcesConfig.from_app_config(self.config)
+        )
+        self._register_abstract_singleton(
+            UserDefinedFileSources, UserDefinedFileSourcesImpl  # type: ignore[type-abstract]  # https://github.com/python/mypy/issues/4717
+        )
+        templates = ConfiguredFileSourceTemplates.from_app_config(self.config, vault_configured=vault_configured)
+        self.file_source_templates = self._register_singleton(ConfiguredFileSourceTemplates, templates)
+        configured_file_source_conf: ConfiguredFileSourcesConf = ConfiguredFileSourcesConf.from_app_config(self.config)
+        file_sources = ConfiguredFileSources(
+            file_sources_config,
+            configured_file_source_conf,
+            load_stock_plugins=True,
+            plugin_loader=file_source_plugin_loader,
         )
+        self.file_sources = self._register_singleton(ConfiguredFileSources, file_sources)
+        self._register_singleton(FileSourceInstancesManager)
 
         # Load security policy.
         self.security_agent = self.model.security_agent
diff --git a/lib/galaxy/app_unittest_utils/galaxy_mock.py b/lib/galaxy/app_unittest_utils/galaxy_mock.py
index d50842158020..7fc6c2200e8d 100644
--- a/lib/galaxy/app_unittest_utils/galaxy_mock.py
+++ b/lib/galaxy/app_unittest_utils/galaxy_mock.py
@@ -42,6 +42,11 @@
     GalaxyDataTestConfig,
 )
 from galaxy.security import idencoding
+from galaxy.security.vault import (
+    UserVaultWrapper,
+    Vault,
+    VaultFactory,
+)
 from galaxy.short_term_storage import (
     ShortTermStorageAllocator,
     ShortTermStorageConfiguration,
@@ -171,6 +176,18 @@ def wait_for_toolbox_reload(self, toolbox):
     def reindex_tool_search(self) -> None:
         raise NotImplementedError
 
+    def setup_test_vault(self):
+        config = {
+            "encryption_keys": [
+                "5RrT94ji178vQwha7TAmEix7DojtsLlxVz8Ef17KWgg=",
+                "iNdXd7tRjLnSqRHxuhqQ98GTLU8HUbd5_Xx38iF8nZ0=",
+                "IK83IXhE4_7W7xCFEtD9op0BAs11pJqYN236Spppp7g=",
+            ],
+        }
+        vault = VaultFactory.from_vault_type(self, "database", config)
+        self[Vault] = vault
+        self.vault = vault
+
 
 class MockLock:
     def __enter__(self):
@@ -352,6 +369,11 @@ def fill_template(self, filename, template_lookup=None, **kwargs):
         kwargs.update(h=MockTemplateHelpers())
         return template.render(**kwargs)
 
+    @property
+    def user_vault(self):
+        """Provide access to a user's personal vault."""
+        return UserVaultWrapper(self.app.vault, self.user)
+
 
 class MockVisualizationsRegistry:
     BUILT_IN_VISUALIZATIONS = ["trackster"]
diff --git a/lib/galaxy/config/schemas/config_schema.yml b/lib/galaxy/config/schemas/config_schema.yml
index 1f5c79c1ffbd..38a626c8f6b5 100644
--- a/lib/galaxy/config/schemas/config_schema.yml
+++ b/lib/galaxy/config/schemas/config_schema.yml
@@ -566,6 +566,21 @@ mapping:
         desc: |
           Configured Object Store templates embedded into Galaxy's config.
 
+      file_source_templates_config_file:
+        type: str
+        default: file_source_templates.yml
+        path_resolves_to: config_dir
+        required: false
+        desc: |
+          Configured user file source templates configuration file.
+
+      file_source_templates:
+        type: seq
+        sequence:
+          - type: any
+        desc: |
+          Configured user file source templates embedded into Galaxy's config.
+
       user_object_store_index_by:
         type: str
         default: 'uuid'
diff --git a/lib/galaxy/files/__init__.py b/lib/galaxy/files/__init__.py
index f58f158ee38c..f459720368b3 100644
--- a/lib/galaxy/files/__init__.py
+++ b/lib/galaxy/files/__init__.py
@@ -7,6 +7,7 @@
     List,
     NamedTuple,
     Optional,
+    Protocol,
     Set,
 )
 
@@ -21,6 +22,7 @@
     plugin_source_from_dict,
     plugin_source_from_path,
     PluginConfigSource,
+    PluginConfigsT,
 )
 from .plugins import (
     FileSourcePluginLoader,
@@ -44,25 +46,107 @@ class NoMatchingFileSource(Exception):
     pass
 
 
+class UserDefinedFileSources(Protocol):
+    """Entry-point for Galaxy to inject user-defined object stores.
+
+    Supplied object of this class is used to write out concrete
+    description of file sources when serializing all file sources
+    available to a user.
+    """
+
+    def validate_uri_root(self, uri: str, user_context: "ProvidesUserFileSourcesUserContext") -> None:
+        pass
+
+    def find_best_match(self, url: str) -> Optional[FileSourceScore]:
+        pass
+
+    def user_file_sources_to_dicts(
+        self,
+        for_serialization: bool,
+        user_context: "FileSourceDictifiable",
+        browsable_only: Optional[bool] = False,
+        include_kind: Optional[Set[PluginKind]] = None,
+        exclude_kind: Optional[Set[PluginKind]] = None,
+    ) -> List[FilesSourceProperties]:
+        """Write out user file sources as list of config dictionaries."""
+        # config_dicts: List[FilesSourceProperties] = []
+        # for file_source in self.user_file_sources():
+        #     as_dict = file_source.to_dict(for_serialization=for_serialization, user_context=user_context)
+        #     config_dicts.append(as_dict)
+        # return config_dicts
+
+
+class NullUserDefinedFileSources(UserDefinedFileSources):
+
+    def validate_uri_root(self, uri: str, user_context: "ProvidesUserFileSourcesUserContext") -> None:
+        return None
+
+    def find_best_match(self, url: str) -> Optional[FileSourceScore]:
+        return None
+
+    def user_file_sources_to_dicts(
+        self,
+        for_serialization: bool,
+        user_context: "FileSourceDictifiable",
+        browsable_only: Optional[bool] = False,
+        include_kind: Optional[Set[PluginKind]] = None,
+        exclude_kind: Optional[Set[PluginKind]] = None,
+    ) -> List[FilesSourceProperties]:
+        return []
+
+
+def _ensure_user_defined_file_sources(
+    user_defined_file_sources: Optional[UserDefinedFileSources] = None,
+) -> UserDefinedFileSources:
+    if user_defined_file_sources is not None:
+        return user_defined_file_sources
+    else:
+        return NullUserDefinedFileSources()
+
+
+class ConfiguredFileSourcesConf:
+    conf_dict: Optional[PluginConfigsT]
+    conf_file: Optional[str]
+
+    def __init__(self, conf_dict: Optional[PluginConfigsT] = None, conf_file: Optional[str] = None):
+        self.conf_dict = conf_dict
+        self.conf_file = conf_file
+
+    @staticmethod
+    def from_app_config(config):
+        config_file = config.file_sources_config_file
+        config_dict = None
+        if not config_file or not os.path.exists(config_file):
+            config_file = None
+            config_dict = config.file_sources
+        return ConfiguredFileSourcesConf(config_dict, config_file)
+
+
 class ConfiguredFileSources:
     """Load plugins and resolve Galaxy URIs to FileSource objects."""
 
     _file_sources: List[BaseFilesSource]
+    _plugin_loader: FileSourcePluginLoader
+    _user_defined_file_sources: UserDefinedFileSources
 
     def __init__(
         self,
         file_sources_config: FileSourcePluginsConfig,
-        conf_file=None,
-        conf_dict=None,
-        load_stock_plugins=False,
+        configured_file_source_conf: Optional[ConfiguredFileSourcesConf] = None,
+        load_stock_plugins: bool = False,
+        plugin_loader: Optional[FileSourcePluginLoader] = None,
+        user_defined_file_sources: Optional[UserDefinedFileSources] = None,
     ):
         self._file_sources_config = file_sources_config
-        self._plugin_loader = FileSourcePluginLoader()
+        self._plugin_loader = plugin_loader or FileSourcePluginLoader()
+        self._user_defined_file_sources = _ensure_user_defined_file_sources(user_defined_file_sources)
         file_sources: List[BaseFilesSource] = []
-        if conf_file is not None:
-            file_sources = self._load_plugins_from_file(conf_file)
-        elif conf_dict is not None:
-            plugin_source = plugin_source_from_dict(conf_dict)
+        if configured_file_source_conf is None:
+            configured_file_source_conf = ConfiguredFileSourcesConf(conf_dict=[])
+        if configured_file_source_conf.conf_file is not None:
+            file_sources = self._load_plugins_from_file(configured_file_source_conf.conf_file)
+        elif configured_file_source_conf.conf_dict is not None:
+            plugin_source = plugin_source_from_dict(configured_file_source_conf.conf_dict)
             file_sources = self._parse_plugin_source(plugin_source)
         else:
             file_sources = []
@@ -152,6 +236,7 @@ def validate_uri_root(self, uri: str, user_context: "ProvidesUserFileSourcesUser
                 raise exceptions.ObjectNotFound(
                     "Your FTP directory does not exist, attempting to upload files to it may cause it to be created."
                 )
+        self._user_defined_file_sources.validate_uri_root(uri, user_context)
 
     def looks_like_uri(self, path_or_uri):
         # is this string a URI this object understands how to realize
@@ -181,6 +266,16 @@ def plugins_to_dict(
                 continue
             el = file_source.to_dict(for_serialization=for_serialization, user_context=user_context)
             rval.append(el)
+        if user_context:
+            rval.extend(
+                self._user_defined_file_sources.user_file_sources_to_dicts(
+                    for_serialization,
+                    user_context,
+                    browsable_only=browsable_only,
+                    include_kind=include_kind,
+                    exclude_kind=exclude_kind,
+                )
+            )
         return rval
 
     def to_dict(
@@ -191,18 +286,6 @@ def to_dict(
             "config": self._file_sources_config.to_dict(),
         }
 
-    @staticmethod
-    def from_app_config(config):
-        config_file = config.file_sources_config_file
-        config_dict = None
-        if not config_file or not os.path.exists(config_file):
-            config_file = None
-            config_dict = config.file_sources
-        file_sources_config = FileSourcePluginsConfig.from_app_config(config)
-        return ConfiguredFileSources(
-            file_sources_config, conf_file=config_file, conf_dict=config_dict, load_stock_plugins=True
-        )
-
     @staticmethod
     def from_dict(as_dict, load_stock_plugins=False):
         if as_dict is not None:
@@ -212,8 +295,9 @@ def from_dict(as_dict, load_stock_plugins=False):
         else:
             sources_as_dict = []
             file_sources_config = FileSourcePluginsConfig()
+        configured_file_sources_conf = ConfiguredFileSourcesConf(conf_dict=sources_as_dict)
         return ConfiguredFileSources(
-            file_sources_config, conf_dict=sources_as_dict, load_stock_plugins=load_stock_plugins
+            file_sources_config, configured_file_sources_conf, load_stock_plugins=load_stock_plugins
         )
 
 
@@ -221,7 +305,7 @@ class NullConfiguredFileSources(ConfiguredFileSources):
     def __init__(
         self,
     ):
-        super().__init__(FileSourcePluginsConfig())
+        super().__init__(FileSourcePluginsConfig(), ConfiguredFileSourcesConf(conf_dict=[]))
 
 
 class FileSourceDictifiable(Dictifiable):
@@ -233,6 +317,10 @@ def to_dict(self, view="collection", value_mapper=None):
         rval["group_names"] = list(self.group_names)
         return rval
 
+    @property
+    def username(self) -> Optional[str]:
+        raise NotImplementedError
+
     @property
     def role_names(self) -> Set[str]:
         raise NotImplementedError
@@ -259,7 +347,7 @@ def email(self):
         return user and user.email
 
     @property
-    def username(self):
+    def username(self) -> Optional[str]:
         user = self.trans.user
         return user and user.username
 
@@ -315,7 +403,7 @@ def email(self):
         return self._kwd.get("email")
 
     @property
-    def username(self):
+    def username(self) -> Optional[str]:
         return self._kwd.get("username")
 
     @property
diff --git a/lib/galaxy/files/plugins.py b/lib/galaxy/files/plugins.py
index 27f5f2de96d2..4f9c182f846a 100644
--- a/lib/galaxy/files/plugins.py
+++ b/lib/galaxy/files/plugins.py
@@ -1,6 +1,8 @@
 from typing import (
+    cast,
     List,
     Optional,
+    Type,
     TYPE_CHECKING,
 )
 
@@ -86,6 +88,9 @@ def _file_source_plugins_dict(self):
 
         return plugins_dict(galaxy.files.sources, "plugin_type")
 
+    def get_plugin_type_class(self, plugin_type: str) -> Type["BaseFilesSource"]:
+        return cast(Type["BaseFilesSource"], self._plugin_classes[plugin_type])
+
     def load_plugins(
         self, plugin_source: PluginConfigSource, file_source_plugin_config: FileSourcePluginsConfig
     ) -> List["BaseFilesSource"]:
diff --git a/lib/galaxy/files/sources/__init__.py b/lib/galaxy/files/sources/__init__.py
index c0a79e7abe67..da8791d69744 100644
--- a/lib/galaxy/files/sources/__init__.py
+++ b/lib/galaxy/files/sources/__init__.py
@@ -11,6 +11,7 @@
     ClassVar,
     Optional,
     Set,
+    Type,
 )
 
 from typing_extensions import (
@@ -271,18 +272,22 @@ def get_browsable(self) -> bool:
         """Return true if the filesource implements the SupportsBrowsing interface."""
 
 
+def file_source_type_is_browsable(target_type: Type["BaseFilesSource"]) -> bool:
+    # Check whether the list method has been overridden
+    return target_type.list != BaseFilesSource.list or target_type._list != BaseFilesSource._list
+
+
 class BaseFilesSource(FilesSource):
     plugin_kind: ClassVar[PluginKind] = PluginKind.rfs  # Remote File Source by default, override in subclasses
 
     def get_browsable(self) -> bool:
-        # Check whether the list method has been overridden
-        return type(self).list != BaseFilesSource.list or type(self)._list != BaseFilesSource._list
+        return file_source_type_is_browsable(type(self))
 
     def get_prefix(self) -> Optional[str]:
         return self.id
 
     def get_scheme(self) -> str:
-        return "gxfiles"
+        return self.scheme or "gxfiles"
 
     def get_writable(self) -> bool:
         return self.writable
diff --git a/lib/galaxy/files/templates/__init__.py b/lib/galaxy/files/templates/__init__.py
new file mode 100644
index 000000000000..aa8faf85fd30
--- /dev/null
+++ b/lib/galaxy/files/templates/__init__.py
@@ -0,0 +1,17 @@
+from .manager import ConfiguredFileSourceTemplates
+from .models import (
+    FileSourceConfiguration,
+    FileSourceTemplate,
+    FileSourceTemplateSummaries,
+    FileSourceTemplateType,
+    template_to_configuration,
+)
+
+__all__ = (
+    "ConfiguredFileSourceTemplates",
+    "FileSourceConfiguration",
+    "FileSourceTemplate",
+    "FileSourceTemplateSummaries",
+    "FileSourceTemplateType",
+    "template_to_configuration",
+)
diff --git a/lib/galaxy/files/templates/manager.py b/lib/galaxy/files/templates/manager.py
new file mode 100644
index 000000000000..c757e6482014
--- /dev/null
+++ b/lib/galaxy/files/templates/manager.py
@@ -0,0 +1,86 @@
+import os
+from typing import (
+    List,
+    Optional,
+    Protocol,
+)
+
+from yaml import safe_load
+
+from galaxy.util.config_templates import (
+    apply_syntactic_sugar,
+    find_template,
+    find_template_by,
+    InstanceDefinition,
+    RawTemplateConfig,
+    TemplateReference,
+    validate_secrets_and_variables,
+    verify_vault_configured_if_uses_secrets,
+)
+from .models import (
+    FileSourceTemplate,
+    FileSourceTemplateCatalog,
+    FileSourceTemplateSummaries,
+)
+
+SECRETS_NEED_VAULT_MESSAGE = "The file source templates configuration can not be used - a Galaxy vault must be configured for templates that use secrets - please set the vault_config_file configuration option to point at a valid vault configuration."
+
+
+class AppConfigProtocol(Protocol):
+    file_source_templates: Optional[List[RawTemplateConfig]]
+    file_source_templates_config_file: Optional[str]
+
+
+class ConfiguredFileSourceTemplates:
+    catalog: FileSourceTemplateCatalog
+
+    def __init__(self, catalog: FileSourceTemplateCatalog):
+        self.catalog = catalog
+
+    @staticmethod
+    def from_app_config(config: AppConfigProtocol, vault_configured=None) -> "ConfiguredFileSourceTemplates":
+        raw_config = config.file_source_templates
+        if raw_config is None:
+            config_file = config.file_source_templates_config_file
+            if config_file and os.path.exists(config_file):
+                with open(config_file) as f:
+                    raw_config = safe_load(f)
+        if raw_config is None:
+            raw_config = []
+        catalog = raw_config_to_catalog(raw_config)
+        verify_vault_configured_if_uses_secrets(
+            catalog,
+            vault_configured,
+            SECRETS_NEED_VAULT_MESSAGE,
+        )
+        templates = ConfiguredFileSourceTemplates(catalog)
+        return templates
+
+    @property
+    def summaries(self) -> FileSourceTemplateSummaries:
+        templates = self.catalog.root
+        summaries = []
+        for template in templates:
+            template_dict = template.model_dump()
+            configuration = template_dict.pop("configuration")
+            object_store_type = configuration["type"]
+            template_dict["type"] = object_store_type
+            summaries.append(template_dict)
+        return FileSourceTemplateSummaries.model_validate(summaries)
+
+    def find_template(self, instance_reference: TemplateReference) -> FileSourceTemplate:
+        """Find the corresponding template and throw ObjectNotFound if not available."""
+        return find_template(self.catalog.root, instance_reference, "file source")
+
+    def find_template_by(self, template_id: str, template_version: int) -> FileSourceTemplate:
+        return find_template_by(self.catalog.root, template_id, template_version, "file source")
+
+    def validate(self, instance: InstanceDefinition):
+        template = self.find_template(instance)
+        validate_secrets_and_variables(instance, template)
+        # TODO: validate no extra variables
+
+
+def raw_config_to_catalog(raw_config: List[RawTemplateConfig]) -> FileSourceTemplateCatalog:
+    effective_root = apply_syntactic_sugar(raw_config)
+    return FileSourceTemplateCatalog.model_validate(effective_root)
diff --git a/lib/galaxy/files/templates/models.py b/lib/galaxy/files/templates/models.py
new file mode 100644
index 000000000000..63eb0b17c954
--- /dev/null
+++ b/lib/galaxy/files/templates/models.py
@@ -0,0 +1,125 @@
+from typing import (
+    Any,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Type,
+    Union,
+)
+
+from pydantic import RootModel
+
+from galaxy.util.config_templates import (
+    expand_raw_config,
+    MarkdownContent,
+    StrictModel,
+    TemplateExpansion,
+    TemplateSecret,
+    TemplateVariable,
+    TemplateVariableValueType,
+)
+
+FileSourceTemplateType = Literal["posix", "s3fs"]
+
+
+class PosixFileSourceTemplateConfiguration(StrictModel):
+    type: Literal["posix"]
+    root: Union[str, TemplateExpansion]
+
+
+class PosixFileSourceConfiguration(StrictModel):
+    type: Literal["posix"]
+    root: str
+
+
+class S3FSFileSourceTemplateConfiguration(StrictModel):
+    type: Literal["s3fs"]
+    endpoint_url: Optional[Union[str, TemplateExpansion]] = None
+    anon: Optional[Union[bool, TemplateExpansion]] = False
+    secret: Optional[Union[str, TemplateExpansion]] = None
+    key: Optional[Union[str, TemplateExpansion]] = None
+    bucket: Optional[Union[str, TemplateExpansion]] = None
+
+
+class S3FSFileSourceConfiguration(StrictModel):
+    type: Literal["s3fs"]
+    endpoint_url: Optional[str] = None
+    anon: Optional[bool] = False
+    secret: Optional[str] = None
+    key: Optional[str] = None
+    bucket: Optional[str] = None
+
+
+FileSourceTemplateConfiguration = Union[
+    PosixFileSourceTemplateConfiguration,
+    S3FSFileSourceTemplateConfiguration,
+]
+FileSourceConfiguration = Union[
+    PosixFileSourceConfiguration,
+    S3FSFileSourceConfiguration,
+]
+
+
+class FileSourceTemplateBase(StrictModel):
+    """Version of FileSourceTemplate we can send to the UI/API.
+
+    The configuration key in the child type may have secretes
+    and shouldn't be exposed over the API - at least to non-admins.
+    """
+
+    id: str
+    name: Optional[str]
+    description: Optional[MarkdownContent]
+    # The UI should just show the most recent version but allow
+    # admins to define newer versions with new parameterizations
+    # and keep old versions in template catalog for backward compatibility
+    # for users with existing stores of that template.
+    version: int = 0
+    # Like with multiple versions, allow admins to deprecate a
+    # template by hiding but keep it in the catalog for backward
+    # compatibility for users with existing stores of that template.
+    hidden: bool = False
+    variables: Optional[List[TemplateVariable]] = None
+    secrets: Optional[List[TemplateSecret]] = None
+
+
+class FileSourceTemplateSummary(FileSourceTemplateBase):
+    type: FileSourceTemplateType
+
+
+class FileSourceTemplate(FileSourceTemplateBase):
+    configuration: FileSourceTemplateConfiguration
+
+
+FileSourceTemplateCatalog = RootModel[List[FileSourceTemplate]]
+
+
+class FileSourceTemplateSummaries(RootModel):
+    root: List[FileSourceTemplateSummary]
+
+
+def template_to_configuration(
+    template: FileSourceTemplate,
+    variables: Dict[str, TemplateVariableValueType],
+    secrets: Dict[str, str],
+    user_details: Dict[str, Any],
+) -> FileSourceConfiguration:
+    configuration_template = template.configuration
+    raw_config = expand_raw_config(configuration_template, variables, secrets, user_details)
+    return to_configuration_object(raw_config)
+
+
+TypesToConfigurationClasses: Dict[FileSourceTemplateType, Type[FileSourceConfiguration]] = {
+    "posix": PosixFileSourceConfiguration,
+    "s3fs": S3FSFileSourceConfiguration,
+}
+
+
+def to_configuration_object(configuration_dict: Dict[str, Any]) -> FileSourceConfiguration:
+    if "type" not in configuration_dict:
+        raise KeyError("Configuration objects require a file source 'type' key, none found.")
+    object_store_type = configuration_dict["type"]
+    if object_store_type not in TypesToConfigurationClasses:
+        raise ValueError(f"Unknown file source type found in raw configuration dictionary ({object_store_type}).")
+    return TypesToConfigurationClasses[object_store_type](**configuration_dict)
diff --git a/lib/galaxy/files/unittest_utils/__init__.py b/lib/galaxy/files/unittest_utils/__init__.py
index c8ebbc13a5b3..e04da53aabd9 100644
--- a/lib/galaxy/files/unittest_utils/__init__.py
+++ b/lib/galaxy/files/unittest_utils/__init__.py
@@ -2,13 +2,16 @@
 import tempfile
 from typing import Tuple
 
-from galaxy.files import ConfiguredFileSources
+from galaxy.files import (
+    ConfiguredFileSources,
+    ConfiguredFileSourcesConf,
+)
 from galaxy.files.plugins import FileSourcePluginsConfig
 
 
 class TestConfiguredFileSources(ConfiguredFileSources):
     def __init__(self, file_sources_config: FileSourcePluginsConfig, conf_dict: dict, test_root: str):
-        super().__init__(file_sources_config, conf_dict=conf_dict)
+        super().__init__(file_sources_config, ConfiguredFileSourcesConf(conf_dict=conf_dict))
         self.test_root = test_root
 
 
diff --git a/lib/galaxy/managers/file_source_instances.py b/lib/galaxy/managers/file_source_instances.py
new file mode 100644
index 000000000000..48742c24b71c
--- /dev/null
+++ b/lib/galaxy/managers/file_source_instances.py
@@ -0,0 +1,420 @@
+import logging
+from typing import (
+    Any,
+    cast,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Set,
+    Union,
+)
+from uuid import uuid4
+
+from pydantic import BaseModel
+
+from galaxy.exceptions import (
+    ItemOwnershipException,
+    RequestParameterInvalidException,
+    RequestParameterMissingException,
+)
+from galaxy.files import (
+    FileSourceDictifiable,
+    FileSourceScore,
+    ProvidesUserFileSourcesUserContext,
+    UserDefinedFileSources,
+)
+from galaxy.files.plugins import (
+    FileSourcePluginLoader,
+    FileSourcePluginsConfig,
+)
+from galaxy.files.sources import (
+    BaseFilesSource,
+    file_source_type_is_browsable,
+    FilesSourceProperties,
+    PluginKind,
+)
+from galaxy.files.templates import (
+    ConfiguredFileSourceTemplates,
+    FileSourceConfiguration,
+    FileSourceTemplateSummaries,
+    FileSourceTemplateType,
+)
+from galaxy.managers.context import ProvidesUserContext
+from galaxy.model import (
+    OBJECT_STORE_TEMPLATE_CONFIGURATION_VARIABLES_TYPE,
+    User,
+    UserFileSource,
+)
+from galaxy.model.scoped_session import galaxy_scoped_session
+from galaxy.security.vault import Vault
+from galaxy.util.config_templates import TemplateVariableValueType
+from galaxy.util.plugin_config import plugin_source_from_dict
+from .object_store_instances import (
+    CreateInstancePayload,
+    ModifyInstancePayload,
+    recover_secrets,
+    UpdateInstancePayload,
+    UpdateInstanceSecretPayload,
+    UpgradeInstancePayload,
+)
+
+log = logging.getLogger(__name__)
+
+USER_FILE_SOURCES_SCHEME = "gxuserfiles"
+
+
+class UserFileSourceModel(BaseModel):
+    id: Union[str, int]
+    uuid: str
+    uri_root: str
+    name: str
+    description: Optional[str]
+    type: FileSourceTemplateType
+    template_id: str
+    template_version: int
+    variables: Optional[Dict[str, TemplateVariableValueType]]
+    secrets: List[str]
+
+
+class UserDefinedFileSourcesConfig(BaseModel):
+    user_object_store_index_by: Literal["uuid", "id"]
+
+    @staticmethod
+    def from_app_config(config) -> "UserDefinedFileSourcesConfig":
+        user_object_store_index_by = config.user_object_store_index_by
+        assert user_object_store_index_by in ["uuid", "id"]
+        return UserDefinedFileSourcesConfig(user_object_store_index_by=user_object_store_index_by)
+
+
+class FileSourceInstancesManager:
+    _catalog: ConfiguredFileSourceTemplates
+    _sa_session: galaxy_scoped_session
+    _app_vault: Vault
+    _app_config: UserDefinedFileSourcesConfig
+
+    def __init__(
+        self,
+        catalog: ConfiguredFileSourceTemplates,
+        sa_session: galaxy_scoped_session,
+        vault: Vault,
+        app_config: UserDefinedFileSourcesConfig,
+    ):
+        self._catalog = catalog
+        self._sa_session = sa_session
+        self._app_vault = vault
+        self._app_config = app_config
+
+    @property
+    def summaries(self) -> FileSourceTemplateSummaries:
+        return self._catalog.summaries
+
+    def index(self, trans: ProvidesUserContext) -> List[UserFileSourceModel]:
+        stores = self._sa_session.query(UserFileSource).filter(UserFileSource.user_id == trans.user.id).all()
+        return [self._to_model(trans, s) for s in stores]
+
+    def show(self, trans: ProvidesUserContext, id: Union[str, int]) -> UserFileSourceModel:
+        user_file_source = self._get(trans, id)
+        return self._to_model(trans, user_file_source)
+
+    def modify_instance(
+        self, trans: ProvidesUserContext, id: Union[str, int], payload: ModifyInstancePayload
+    ) -> UserFileSourceModel:
+        if isinstance(payload, UpgradeInstancePayload):
+            return self._upgrade_instance(trans, id, payload)
+        elif isinstance(payload, UpdateInstanceSecretPayload):
+            return self._update_instance_secret(trans, id, payload)
+        else:
+            assert isinstance(payload, UpdateInstancePayload)
+            return self._update_instance(trans, id, payload)
+
+    def _upgrade_instance(
+        self, trans: ProvidesUserContext, id: Union[str, int], payload: UpgradeInstancePayload
+    ) -> UserFileSourceModel:
+        persisted_object_store = self._get(trans, id)
+        catalog = self._catalog
+        template = catalog.find_template_by(persisted_object_store.template_id, payload.template_version)
+        persisted_object_store.template_version = template.version
+        persisted_object_store.template_definition = template.model_dump()
+        old_variables = persisted_object_store.template_variables or {}
+        updated_variables = payload.variables
+        actual_variables: OBJECT_STORE_TEMPLATE_CONFIGURATION_VARIABLES_TYPE = {}
+        for variable in template.variables or []:
+            variable_name = variable.name
+            old_value = old_variables.get(variable_name)
+            updated_value = updated_variables.get(variable_name, old_value)
+            if updated_value:
+                actual_variables[variable_name] = updated_value
+
+        persisted_object_store.template_variables = actual_variables
+        old_secrets = persisted_object_store.template_secrets or []
+        new_secrets = payload.secrets
+
+        recorded_secrets = persisted_object_store.template_secrets or []
+
+        user_vault = trans.user_vault
+        upgraded_template_secrets = []
+        for secret in template.secrets or []:
+            secret_name = secret.name
+            upgraded_template_secrets.append(secret_name)
+            if secret_name not in new_secrets and secret_name not in old_secrets:
+                raise RequestParameterMissingException(f"secret {secret_name} not set in supplied request")
+            if secret_name not in new_secrets:
+                # keep old value
+                continue
+
+            secret_value = new_secrets[secret_name]
+            key = persisted_object_store.vault_key(secret_name, self._app_config)
+            user_vault.write_secret(key, secret_value)
+            if secret_name not in recorded_secrets:
+                recorded_secrets.append(secret_name)
+
+        secrets_to_delete: List[str] = []
+        for recorded_secret in recorded_secrets:
+            if recorded_secret not in upgraded_template_secrets:
+                key = persisted_object_store.vault_key(recorded_secret, self._app_config)
+                log.info(f"deleting {key} from user vault")
+                user_vault.delete_secret(key)
+                secrets_to_delete.append(recorded_secret)
+
+        for secret_to_delete in secrets_to_delete:
+            recorded_secrets.remove(secret_to_delete)
+
+        persisted_object_store.template_secrets = recorded_secrets
+        self._save(persisted_object_store)
+        rval = self._to_model(trans, persisted_object_store)
+        return rval
+
+    def _update_instance(
+        self, trans: ProvidesUserContext, id: Union[str, int], payload: UpdateInstancePayload
+    ) -> UserFileSourceModel:
+        persisted_file_source = self._get(trans, id)
+        if payload.name is not None:
+            persisted_file_source.name = payload.name
+        if payload.description is not None:
+            persisted_file_source.description = payload.description
+        if payload.variables is not None:
+            # maybe just record the valid variables according to template like in upgrade
+            persisted_file_source.template_variables = payload.variables
+        self._save(persisted_file_source)
+        return self._to_model(trans, persisted_file_source)
+
+    def _update_instance_secret(
+        self, trans: ProvidesUserContext, id: Union[str, int], payload: UpdateInstanceSecretPayload
+    ) -> UserFileSourceModel:
+        persisted_file_source = self._get(trans, id)
+        user_vault = trans.user_vault
+        key = persisted_file_source.vault_key(payload.secret_name, self._app_config)
+        user_vault.write_secret(key, payload.secret_value)
+        return self._to_model(trans, persisted_file_source)
+
+    def create_instance(self, trans: ProvidesUserContext, payload: CreateInstancePayload) -> UserFileSourceModel:
+        catalog = self._catalog
+        catalog.validate(payload)
+        template = catalog.find_template(payload)
+        assert template
+        user_vault = trans.user_vault
+        persisted_file_source = UserFileSource()
+        persisted_file_source.user_id = trans.user.id
+        assert persisted_file_source.user_id
+        persisted_file_source.uuid = uuid4().hex
+        persisted_file_source.template_definition = template.model_dump()
+        persisted_file_source.template_id = template.id
+        persisted_file_source.template_version = template.version
+        persisted_file_source.template_variables = payload.variables
+        persisted_file_source.name = payload.name
+        persisted_file_source.description = payload.description
+        self._save(persisted_file_source)
+
+        # the exception handling below will cleanup object stores that cannot be
+        # finalized with a successful secret setting but it might be worth considering
+        # something more robust. Two ideas would be to set a uuid on the persisted_file_source
+        # and key the secrets on that instead of the of the ID (but this raises the question
+        # are unused secrets in the vault maybe even worse than broken db objects) or
+        # set a state on the DB objects and with INITIAL and ACTIVE states. State
+        # idea might be nice because then we could add INACTIVE state that would prevent
+        # new data from being added but still allow access.
+        recorded_secrets = []
+        try:
+            for secret, value in payload.secrets.items():
+                key = persisted_file_source.vault_key(secret, self._app_config)
+                user_vault.write_secret(key, value)
+                recorded_secrets.append(secret)
+        except Exception:
+            self._sa_session.delete(persisted_file_source)
+            raise
+        persisted_file_source.template_secrets = recorded_secrets
+        self._save(persisted_file_source)
+        return self._to_model(trans, persisted_file_source)
+
+    def _index_filter(self, id: Union[str, int]):
+        index_by = self._app_config.user_object_store_index_by
+        index_filter: Any
+        if index_by == "id":
+            id_as_int = int(id)
+            index_filter = UserFileSource.__table__.c.id == id_as_int
+        else:
+            id_as_str = str(id)
+            index_filter = UserFileSource.__table__.c.uuid == id_as_str
+        return index_filter
+
+    def _get(self, trans: ProvidesUserContext, id: Union[str, int]) -> UserFileSource:
+        filter = self._index_filter(id)
+        user_file_source = self._sa_session.query(UserFileSource).filter(filter).one_or_none()
+        if user_file_source is None:
+            raise RequestParameterInvalidException(f"Failed to fetch object store for id {id}")
+        if user_file_source.user != trans.user:
+            raise ItemOwnershipException()
+        return user_file_source
+
+    def _save(self, user_file_source: UserFileSource) -> None:
+        self._sa_session.add(user_file_source)
+        self._sa_session.flush([user_file_source])
+        self._sa_session.commit()
+
+    def _to_model(self, trans, persisted_file_source: UserFileSource) -> UserFileSourceModel:
+        file_source_type = persisted_file_source.template.configuration.type
+        # These shouldn't be null but sometimes can be?
+        secrets = persisted_file_source.template_secrets or []
+        index_by = self._app_config.user_object_store_index_by
+        if index_by == "id":
+            ufs_id = str(persisted_file_source.id)
+            response_id = persisted_file_source.id
+        else:
+            ufs_id = str(persisted_file_source.uuid)
+            response_id = ufs_id
+        uri_root = f"{USER_FILE_SOURCES_SCHEME}://{ufs_id}"
+        return UserFileSourceModel(
+            id=response_id,
+            uuid=str(persisted_file_source.uuid),
+            uri_root=uri_root,
+            type=file_source_type,
+            template_id=persisted_file_source.template_id,
+            template_version=persisted_file_source.template_version,
+            variables=persisted_file_source.template_variables,
+            secrets=secrets,
+            name=persisted_file_source.name,
+            description=persisted_file_source.description,
+        )
+
+
+class UserDefinedFileSourcesImpl(UserDefinedFileSources):
+    _sa_session: galaxy_scoped_session
+    _app_config: UserDefinedFileSourcesConfig
+    _file_sources_config: FileSourcePluginsConfig
+    _plugin_loader: FileSourcePluginLoader
+    _app_vault: Vault
+
+    def __init__(
+        self,
+        sa_session: galaxy_scoped_session,
+        app_config: UserDefinedFileSourcesConfig,
+        file_sources_config: FileSourcePluginsConfig,
+        plugin_loader: FileSourcePluginLoader,
+        vault: Vault,
+    ):
+        self._sa_session = sa_session
+        self._app_config = app_config
+        self._plugin_loader = plugin_loader
+        self._file_sources_config = file_sources_config
+        self._app_vault = vault
+
+    def _user_file_source(self, uri: str) -> Optional[UserFileSource]:
+        if "://" not in uri:
+            return None
+        uri_scheme, uri_root = uri.split("://", 1)
+        if uri_scheme != USER_FILE_SOURCES_SCHEME:
+            return None
+        index_by = self._app_config.user_object_store_index_by
+        index_filter: Any
+        if index_by == "id":
+            index_filter = UserFileSource.__table__.c.id == uri_root
+        else:
+            index_filter = UserFileSource.__table__.c.uuid == uri_root
+
+        user_object_store: UserFileSource = self._sa_session.query(UserFileSource).filter(index_filter).one()
+        return user_object_store
+
+    def _file_source_properties_from_uri(self, uri: str) -> Optional[FilesSourceProperties]:
+        user_file_source = self._user_file_source(uri)
+        if not user_file_source:
+            return None
+        return self._file_source_properties(user_file_source)
+
+    def _file_source_properties(self, user_file_source: UserFileSource) -> FilesSourceProperties:
+        secrets = recover_secrets(user_file_source, self._app_vault, self._app_config)
+        file_source_configuration: FileSourceConfiguration = user_file_source.file_source_configuration(secrets=secrets)
+        file_source_properties = cast(FilesSourceProperties, file_source_configuration.model_dump())
+        file_source_properties["label"] = user_file_source.name
+        file_source_properties["doc"] = user_file_source.description
+        file_source_properties["id"] = f"{user_file_source.uuid}"
+        file_source_properties["scheme"] = USER_FILE_SOURCES_SCHEME
+        # when should this be false - I think it can depend on the plugin type
+        # no reason to prevent users from writing to their own file sources?
+        file_source_properties["writable"] = True
+        return file_source_properties
+
+    def validate_uri_root(self, uri: str, user_context: "ProvidesUserFileSourcesUserContext") -> None:
+        user_object_store = self._user_file_source(uri)
+        if not user_object_store:
+            return
+        if user_object_store.user.username != user_context.username:
+            raise ItemOwnershipException("Your Galaxy user does not have access to the requested resource.")
+
+    def find_best_match(self, url: str) -> Optional[FileSourceScore]:
+        files_source_properties = self._file_source_properties_from_uri(url)
+        if files_source_properties is None:
+            return None
+        file_source = self._file_source(files_source_properties)
+        return FileSourceScore(file_source, len(url))
+
+    def _file_source(self, files_source_properties: FilesSourceProperties) -> BaseFilesSource:
+        plugin_source = plugin_source_from_dict([cast(Dict[str, Any], files_source_properties)])
+        file_source = self._plugin_loader.load_plugins(
+            plugin_source,
+            self._file_sources_config,
+        )[0]
+        return file_source
+
+    def _all_user_file_source_properties(self, user_context: FileSourceDictifiable) -> List[FilesSourceProperties]:
+        username_filter = User.__table__.c.username == user_context.username
+        user: User = self._sa_session.query(User).filter(username_filter).one()
+        all_file_source_properties: List[FilesSourceProperties] = []
+        for user_file_source in user.file_sources:
+            files_source_properties = self._file_source_properties(user_file_source)
+            all_file_source_properties.append(files_source_properties)
+        return all_file_source_properties
+
+    def user_file_sources_to_dicts(
+        self,
+        for_serialization: bool,
+        user_context: FileSourceDictifiable,
+        browsable_only: Optional[bool] = False,
+        include_kind: Optional[Set[PluginKind]] = None,
+        exclude_kind: Optional[Set[PluginKind]] = None,
+    ) -> List[FilesSourceProperties]:
+        """Write out user file sources as list of config dictionaries."""
+        as_dicts = []
+        for files_source_properties in self._all_user_file_source_properties(user_context):
+            plugin_kind = PluginKind.rfs
+            if include_kind and plugin_kind not in include_kind:
+                continue
+            if exclude_kind and plugin_kind in exclude_kind:
+                continue
+            files_source_type = files_source_properties["type"]
+            is_browsable = file_source_type_is_browsable(self._plugin_loader.get_plugin_type_class(files_source_type))
+            if browsable_only and not is_browsable:
+                continue
+            file_source = self._file_source(files_source_properties)
+            as_dicts.append(file_source.to_dict(for_serialization=for_serialization, user_context=user_context))
+        return as_dicts
+
+
+__all__ = (
+    "CreateInstancePayload",
+    "FileSourceInstancesManager",
+    "ModifyInstancePayload",
+    "UpdateInstancePayload",
+    "UpdateInstanceSecretPayload",
+    "UpgradeInstancePayload",
+)
diff --git a/lib/galaxy/managers/object_store_instances.py b/lib/galaxy/managers/object_store_instances.py
index 974ecd0d09e7..8be9808c4830 100644
--- a/lib/galaxy/managers/object_store_instances.py
+++ b/lib/galaxy/managers/object_store_instances.py
@@ -26,9 +26,11 @@
 )
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.model import (
+    HasConfigSecrets,
     OBJECT_STORE_TEMPLATE_CONFIGURATION_VARIABLES_TYPE,
     User,
     UserObjectStore,
+    UsesTemplatesAppConfig,
 )
 from galaxy.model.scoped_session import galaxy_scoped_session
 from galaxy.objectstore import (
@@ -44,11 +46,11 @@
     ObjectStoreTemplateSummaries,
     ObjectStoreTemplateType,
 )
-from galaxy.objectstore.templates.models import ObjectStoreTemplateVariableValueType
 from galaxy.security.vault import (
     UserVaultWrapper,
     Vault,
 )
+from galaxy.util.config_templates import TemplateVariableValueType
 
 log = logging.getLogger(__name__)
 
@@ -58,14 +60,14 @@ class CreateInstancePayload(BaseModel):
     description: Optional[str] = None
     template_id: str
     template_version: int
-    variables: Dict[str, ObjectStoreTemplateVariableValueType]
+    variables: Dict[str, TemplateVariableValueType]
     secrets: Dict[str, str]
 
 
 class UpdateInstancePayload(BaseModel):
     name: Optional[str] = None
     description: Optional[str] = None
-    variables: Optional[Dict[str, ObjectStoreTemplateVariableValueType]] = None
+    variables: Optional[Dict[str, TemplateVariableValueType]] = None
 
 
 class UpdateInstanceSecretPayload(BaseModel):
@@ -75,7 +77,7 @@ class UpdateInstanceSecretPayload(BaseModel):
 
 class UpgradeInstancePayload(BaseModel):
     template_version: int
-    variables: Dict[str, ObjectStoreTemplateVariableValueType]
+    variables: Dict[str, TemplateVariableValueType]
     secrets: Dict[str, str]
 
 
@@ -85,7 +87,7 @@ class UserConcreteObjectStoreModel(ConcreteObjectStoreModel):
     type: ObjectStoreTemplateType
     template_id: str
     template_version: int
-    variables: Optional[Dict[str, ObjectStoreTemplateVariableValueType]]
+    variables: Optional[Dict[str, TemplateVariableValueType]]
     secrets: List[str]
 
 
@@ -160,7 +162,7 @@ def _upgrade_instance(
                 continue
 
             secret_value = new_secrets[secret_name]
-            key = user_vault_key(persisted_object_store, secret_name, self._app_config)
+            key = persisted_object_store.vault_key(secret_name, self._app_config)
             user_vault.write_secret(key, secret_value)
             if secret_name not in recorded_secrets:
                 recorded_secrets.append(secret_name)
@@ -168,7 +170,7 @@ def _upgrade_instance(
         secrets_to_delete: List[str] = []
         for recorded_secret in recorded_secrets:
             if recorded_secret not in upgraded_template_secrets:
-                key = user_vault_key(persisted_object_store, recorded_secret, self._app_config)
+                key = persisted_object_store.vault_key(recorded_secret, self._app_config)
                 log.info(f"deleting {key} from user vault")
                 user_vault.delete_secret(key)
                 secrets_to_delete.append(recorded_secret)
@@ -203,7 +205,7 @@ def _update_instance_secret(
     ) -> UserConcreteObjectStoreModel:
         persisted_object_store = self._get(trans, id)
         user_vault = trans.user_vault
-        key = user_vault_key(persisted_object_store, payload.secret_name, self._app_config)
+        key = persisted_object_store.vault_key(payload.secret_name, self._app_config)
         user_vault.write_secret(key, payload.secret_value)
         return self._to_model(trans, persisted_object_store)
 
@@ -238,7 +240,7 @@ def create_instance(
         recorded_secrets = []
         try:
             for secret, value in payload.secrets.items():
-                key = user_vault_key(persisted_object_store, secret, self._app_config)
+                key = persisted_object_store.vault_key(secret, self._app_config)
                 user_vault.write_secret(key, value)
                 recorded_secrets.append(secret)
         except Exception:
@@ -320,19 +322,8 @@ def _to_model(self, trans, persisted_object_store: UserObjectStore) -> UserConcr
         )
 
 
-def user_vault_key(user_object_store: UserObjectStore, secret: str, app_config: UserObjectStoresAppConfig) -> str:
-    if app_config.user_object_store_index_by == "id":
-        uos_id = str(user_object_store.id)
-    else:
-        uos_id = str(user_object_store.uuid)
-    assert uos_id
-    user_vault_id_prefix = f"object_store_config/{uos_id}"
-    key = f"{user_vault_id_prefix}/{secret}"
-    return key
-
-
 def recover_secrets(
-    user_object_store: UserObjectStore, vault: Vault, app_config: UserObjectStoresAppConfig
+    user_object_store: HasConfigSecrets, vault: Vault, app_config: UsesTemplatesAppConfig
 ) -> Dict[str, str]:
     user: User = user_object_store.user
     user_vault = UserVaultWrapper(vault, user)
@@ -343,9 +334,8 @@ def recover_secrets(
     # persisted.
     persisted_secret_names = user_object_store.template_secrets or []
     for secret in persisted_secret_names:
-        vault_key = user_vault_key(user_object_store, secret, app_config)
+        vault_key = user_object_store.vault_key(secret, app_config)
         secret_value = user_vault.read_secret(vault_key)
-        # assert secret_value
         if secret_value is not None:
             secrets[secret] = secret_value
     return secrets
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 9a4f39a743af..1194920edeb1 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -133,6 +133,11 @@
 import galaxy.model.tags
 import galaxy.security.passwords
 import galaxy.util
+from galaxy.files.templates import (
+    FileSourceConfiguration,
+    FileSourceTemplate,
+    template_to_configuration as file_source_template_to_configuration,
+)
 from galaxy.model.base import (
     ensure_object_added_to_session,
     transaction,
@@ -155,7 +160,7 @@
 from galaxy.objectstore.templates import (
     ObjectStoreConfiguration,
     ObjectStoreTemplate,
-    template_to_configuration,
+    template_to_configuration as object_store_template_to_configuration,
 )
 from galaxy.schema.invocation import (
     InvocationCancellationUserRequest,
@@ -780,6 +785,7 @@ class User(Base, Dictifiable, RepresentById):
         back_populates="user", order_by=lambda: desc(GalaxySession.update_time), cascade_backrefs=False
     )
     object_stores: Mapped[List["UserObjectStore"]] = relationship(back_populates="user")
+    file_sources: Mapped[List["UserFileSource"]] = relationship(back_populates="user")
     quotas: Mapped[List["UserQuotaAssociation"]] = relationship(back_populates="user")
     quota_source_usages: Mapped[List["UserQuotaSourceUsage"]] = relationship(back_populates="user")
     social_auth: Mapped[List["UserAuthnzToken"]] = relationship(back_populates="user")
@@ -10884,8 +10890,29 @@ def __init__(self, name=None, value=None):
         self.value = value
 
 
-class UserObjectStore(Base, RepresentById):
+class UsesTemplatesAppConfig(Protocol):
+    user_object_store_index_by: Literal["uuid", "id"]
+
+
+class HasConfigSecrets(RepresentById):
+    secret_config_type: str
+    template_secrets: Mapped[Optional[OBJECT_STORE_TEMPLATE_CONFIGURATION_SECRET_NAMES_TYPE]]
+    uuid: Mapped[Union[UUID, str]]
+    user: Mapped["User"]
+
+    def vault_key(self, secret: str, app_config: UsesTemplatesAppConfig) -> str:
+        if app_config.user_object_store_index_by == "id":
+            id_str = str(self.id)
+        else:
+            id_str = str(self.uuid)
+        user_vault_id_prefix = f"{self.secret_config_type}/{id_str}"
+        key = f"{user_vault_id_prefix}/{secret}"
+        return key
+
+
+class UserObjectStore(Base, HasConfigSecrets, RepresentById):
     __tablename__ = "user_object_store"
+    secret_config_type = "object_store_config"
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     user_id: Mapped[Optional[int]] = mapped_column(Integer, ForeignKey("galaxy_user.id"), index=True)
@@ -10912,7 +10939,7 @@ class UserObjectStore(Base, RepresentById):
     # Track a list of secrets that were defined for this object store at creation
     template_secrets: Mapped[Optional[OBJECT_STORE_TEMPLATE_CONFIGURATION_SECRET_NAMES_TYPE]] = mapped_column(JSONType)
 
-    user = relationship("User", back_populates="object_stores")
+    user: Mapped["User"] = relationship("User", back_populates="object_stores")
 
     @property
     def template(self) -> ObjectStoreTemplate:
@@ -10926,7 +10953,7 @@ def object_store_configuration(self, secrets: Dict[str, Any]) -> ObjectStoreConf
             "id": user.id,
         }
         variables: OBJECT_STORE_TEMPLATE_CONFIGURATION_VARIABLES_TYPE = self.template_variables or {}
-        return template_to_configuration(
+        return object_store_template_to_configuration(
             self.template,
             variables=variables,
             secrets=secrets,
@@ -10934,6 +10961,57 @@ def object_store_configuration(self, secrets: Dict[str, Any]) -> ObjectStoreConf
         )
 
 
+class UserFileSource(Base, HasConfigSecrets, RepresentById):
+    __tablename__ = "user_file_source"
+    secret_config_type = "file_source_config"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("galaxy_user.id"), index=True)
+    uuid: Mapped[Union[UUID, str]] = mapped_column(UUIDType(), index=True)
+    create_time: Mapped[datetime] = mapped_column(default=now)
+    update_time: Mapped[datetime] = mapped_column(default=now, onupdate=now, index=True)
+    # user specified name of the instance they've created
+    name: Mapped[str] = mapped_column(String(255), index=True)
+    # user specified description of the instance they've created
+    description: Mapped[Optional[str]] = mapped_column(Text)
+    # the template store id
+    template_id: Mapped[str] = mapped_column(String(255), index=True)
+    # the template store version (0, 1, ...)
+    template_version: Mapped[int] = mapped_column(index=True)
+    # Full template from file_sources_templates.yml catalog.
+    # For tools we just store references, so here we could easily just use
+    # the id/version and not record the definition... as the templates change
+    # over time this choice has some big consequences despite being easy to swap
+    # implementations.
+    template_definition: Mapped[Optional[OBJECT_STORE_TEMPLATE_DEFINITION_TYPE]] = mapped_column(JSONType)
+    # Big JSON blob of the variable name -> value mapping defined for the store's
+    # variables by the user.
+    template_variables: Mapped[Optional[OBJECT_STORE_TEMPLATE_CONFIGURATION_VARIABLES_TYPE]] = mapped_column(JSONType)
+    # Track a list of secrets that were defined for this object store at creation
+    template_secrets: Mapped[Optional[OBJECT_STORE_TEMPLATE_CONFIGURATION_SECRET_NAMES_TYPE]] = mapped_column(JSONType)
+
+    user: Mapped["User"] = relationship("User", back_populates="file_sources")
+
+    @property
+    def template(self) -> FileSourceTemplate:
+        return FileSourceTemplate(**self.template_definition or {})
+
+    def file_source_configuration(self, secrets: Dict[str, Any]) -> FileSourceConfiguration:
+        user = self.user
+        user_details = {
+            "username": user.username,
+            "email": user.email,
+            "id": user.id,
+        }
+        variables: OBJECT_STORE_TEMPLATE_CONFIGURATION_VARIABLES_TYPE = self.template_variables or {}
+        return file_source_template_to_configuration(
+            template=self.template,
+            variables=variables,
+            secrets=secrets,
+            user_details=user_details,
+        )
+
+
 class UserAction(Base, RepresentById):
     __tablename__ = "user_action"
 
diff --git a/lib/galaxy/model/migrations/alembic/versions_gxy/c14a3c93d66b_add_user_defined_file_sources.py b/lib/galaxy/model/migrations/alembic/versions_gxy/c14a3c93d66b_add_user_defined_file_sources.py
new file mode 100644
index 000000000000..8e0b57760820
--- /dev/null
+++ b/lib/galaxy/model/migrations/alembic/versions_gxy/c14a3c93d66b_add_user_defined_file_sources.py
@@ -0,0 +1,57 @@
+"""add user defined file sources
+
+Revision ID: c14a3c93d66b
+Revises: c14a3c93d66a
+Create Date: 2023-04-01 17:25:37.553039
+
+"""
+
+from sqlalchemy import (
+    Column,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+)
+
+from galaxy.model.custom_types import (
+    JSONType,
+    UUIDType,
+)
+from galaxy.model.migrations.util import (
+    create_table,
+    drop_table,
+)
+
+# revision identifiers, used by Alembic.
+revision = "c14a3c93d66b"
+down_revision = "c14a3c93d66a"
+branch_labels = None
+depends_on = None
+
+
+# database object names used in this revision
+table_name = "user_file_source"
+
+
+def upgrade():
+    create_table(
+        table_name,
+        Column("id", Integer, primary_key=True),
+        Column("user_id", Integer, ForeignKey("galaxy_user.id"), nullable=False, index=True),
+        Column("uuid", UUIDType, nullable=False, index=True),
+        Column("name", String(255), index=True),
+        Column("description", Text, index=True),
+        Column("create_time", DateTime),
+        Column("update_time", DateTime),
+        Column("template_id", String(255), index=True),
+        Column("template_version", Integer, index=True),
+        Column("template_definition", JSONType),
+        Column("template_variables", JSONType),
+        Column("template_secrets", JSONType),
+    )
+
+
+def downgrade():
+    drop_table(table_name)
diff --git a/lib/galaxy/objectstore/templates/manager.py b/lib/galaxy/objectstore/templates/manager.py
index 901931b956c2..1ae86b60c2ee 100644
--- a/lib/galaxy/objectstore/templates/manager.py
+++ b/lib/galaxy/objectstore/templates/manager.py
@@ -1,7 +1,5 @@
 import os
 from typing import (
-    Any,
-    Dict,
     List,
     Optional,
 )
@@ -9,35 +7,29 @@
 from typing_extensions import Protocol
 from yaml import safe_load
 
-from galaxy.exceptions import (
-    ObjectNotFound,
-    RequestParameterMissingException,
-)
 from galaxy.objectstore.badges import serialize_badges
+from galaxy.util.config_templates import (
+    apply_syntactic_sugar,
+    find_template,
+    find_template_by,
+    InstanceDefinition,
+    RawTemplateConfig,
+    TemplateReference,
+    validate_secrets_and_variables,
+    verify_vault_configured_if_uses_secrets,
+)
 from .models import (
     ObjectStoreTemplate,
     ObjectStoreTemplateCatalog,
     ObjectStoreTemplateSummaries,
 )
 
-RawTemplateConfig = Dict[str, Any]
-
 
 class AppConfigProtocol(Protocol):
     object_store_templates: Optional[List[RawTemplateConfig]]
     object_store_templates_config_file: Optional[str]
 
 
-class TemplateReference(Protocol):
-    template_id: str
-    template_version: int
-
-
-class InstanceDefinition(TemplateReference, Protocol):
-    variables: Dict[str, Any]
-    secrets: Dict[str, str]
-
-
 SECRETS_NEED_VAULT_MESSAGE = "The object store templates configuration can not be used - a Galaxy vault must be configured for templates that use secrets - please set the vault_config_file configuration option to point at a valid vault configuration."
 
 
@@ -57,19 +49,15 @@ def from_app_config(config: AppConfigProtocol, vault_configured=None) -> "Config
                     raw_config = safe_load(f)
         if raw_config is None:
             raw_config = []
-        templates = ConfiguredObjectStoreTemplates(raw_config_to_catalog(raw_config))
-        if vault_configured is False and templates.configuration_uses_secrets:
-            raise Exception(SECRETS_NEED_VAULT_MESSAGE)
+        catalog = raw_config_to_catalog(raw_config)
+        verify_vault_configured_if_uses_secrets(
+            catalog,
+            vault_configured,
+            SECRETS_NEED_VAULT_MESSAGE,
+        )
+        templates = ConfiguredObjectStoreTemplates(catalog)
         return templates
 
-    @property
-    def configuration_uses_secrets(self) -> bool:
-        templates = self.catalog.root
-        for template in templates:
-            if template.secrets and len(template.secrets) > 0:
-                return True
-        return False
-
     @property
     def summaries(self) -> ObjectStoreTemplateSummaries:
         templates = self.catalog.root
@@ -87,55 +75,17 @@ def summaries(self) -> ObjectStoreTemplateSummaries:
 
     def find_template(self, instance_reference: TemplateReference) -> ObjectStoreTemplate:
         """Find the corresponding template and throw ObjectNotFound if not available."""
-        template_id = instance_reference.template_id
-        template_version = instance_reference.template_version
-        return self.find_template_by(template_id, template_version)
+        return find_template(self.catalog.root, instance_reference, "object store")
 
     def find_template_by(self, template_id: str, template_version: int) -> ObjectStoreTemplate:
-        templates = self.catalog.root
-
-        for template in templates:
-            if template.id == template_id and template.version == template_version:
-                return template
-
-        raise ObjectNotFound(
-            f"Could not find a object store template with id {template_id} and version {template_version}"
-        )
+        return find_template_by(self.catalog.root, template_id, template_version, "object store")
 
     def validate(self, instance: InstanceDefinition):
         template = self.find_template(instance)
-        secrets = instance.secrets
-        for template_secret in template.secrets or []:
-            name = template_secret.name
-            if name not in secrets:
-                raise RequestParameterMissingException(f"Must define secret '{name}'")
-        variables = instance.variables
-        for template_variable in template.variables or []:
-            name = template_variable.name
-            if name not in variables:
-                raise RequestParameterMissingException(f"Must define variable '{name}'")
+        validate_secrets_and_variables(instance, template)
         # TODO: validate no extra variables
 
 
 def raw_config_to_catalog(raw_config: List[RawTemplateConfig]) -> ObjectStoreTemplateCatalog:
-    effective_root = _apply_syntactic_sugar(raw_config)
+    effective_root = apply_syntactic_sugar(raw_config)
     return ObjectStoreTemplateCatalog.model_validate(effective_root)
-
-
-def _apply_syntactic_sugar(raw_templates: List[RawTemplateConfig]) -> List[RawTemplateConfig]:
-    templates = []
-    for template in raw_templates:
-        _force_key_to_list(template, "variables")
-        _force_key_to_list(template, "secrets")
-        templates.append(template)
-    return templates
-
-
-def _force_key_to_list(template: RawTemplateConfig, key: str) -> None:
-    value = template.get(key, None)
-    if isinstance(value, dict):
-        value_as_list = []
-        for key_name, key_value in value.items():
-            key_value["name"] = key_name
-            value_as_list.append(key_value)
-        template[key] = value_as_list
diff --git a/lib/galaxy/objectstore/templates/models.py b/lib/galaxy/objectstore/templates/models.py
index 9f225786194a..f32eaf88ac09 100644
--- a/lib/galaxy/objectstore/templates/models.py
+++ b/lib/galaxy/objectstore/templates/models.py
@@ -7,28 +7,26 @@
     Union,
 )
 
-from boltons.iterutils import remap
-from jinja2.nativetypes import NativeEnvironment
-from pydantic import (
-    BaseModel,
-    ConfigDict,
-    RootModel,
-)
+from pydantic import RootModel
 from typing_extensions import Literal
 
 from galaxy.objectstore.badges import (
     BadgeDict,
     StoredBadgeDict,
 )
+from galaxy.util.config_templates import (
+    expand_raw_config,
+    MarkdownContent,
+    StrictModel,
+    TemplateExpansion,
+    TemplateSecret,
+    TemplateVariable,
+    TemplateVariableType,
+    TemplateVariableValueType,
+)
 
-
-class StrictModel(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-
-ObjectStoreTemplateVariableType = Literal["string", "boolean", "integer"]
-ObjectStoreTemplateVariableValueType = Union[str, bool, int]
-TemplateExpansion = str
+ObjectStoreTemplateVariableType = TemplateVariableType
+ObjectStoreTemplateVariableValueType = TemplateVariableValueType
 ObjectStoreTemplateType = Literal["s3", "azure_blob", "disk", "generic_s3"]
 
 
@@ -155,18 +153,10 @@ class GenericS3ObjectStoreConfiguration(StrictModel):
     AzureObjectStoreConfiguration,
     GenericS3ObjectStoreConfiguration,
 ]
-MarkdownContent = str
 
 
-class ObjectStoreTemplateVariable(StrictModel):
-    name: str
-    help: Optional[MarkdownContent]
-    type: ObjectStoreTemplateVariableType
-
-
-class ObjectStoreTemplateSecret(StrictModel):
-    name: str
-    help: Optional[MarkdownContent]
+ObjectStoreTemplateVariable = TemplateVariable
+ObjectStoreTemplateSecret = TemplateSecret
 
 
 class ObjectStoreTemplateBase(StrictModel):
@@ -188,8 +178,8 @@ class ObjectStoreTemplateBase(StrictModel):
     # template by hiding but keep it in the catalog for backward
     # compatibility for users with existing stores of that template.
     hidden: bool = False
-    variables: Optional[List[ObjectStoreTemplateVariable]] = None
-    secrets: Optional[List[ObjectStoreTemplateSecret]] = None
+    variables: Optional[List[TemplateVariable]] = None
+    secrets: Optional[List[TemplateSecret]] = None
 
 
 class ObjectStoreTemplateSummary(ObjectStoreTemplateBase):
@@ -215,20 +205,7 @@ def template_to_configuration(
     user_details: Dict[str, Any],
 ) -> ObjectStoreConfiguration:
     configuration_template = template.configuration
-    template_variables = {
-        "variables": variables,
-        "secrets": secrets,
-        "user": user_details,
-    }
-
-    def expand_template(_, key, value):
-        if isinstance(value, str) and "{{" in value and "}}" in value:
-            # NativeEnvironment preserves Python types
-            template = NativeEnvironment().from_string(value)
-            return key, template.render(**template_variables)
-        return key, value
-
-    raw_config = remap(configuration_template.model_dump(), visit=expand_template)
+    raw_config = expand_raw_config(configuration_template, variables, secrets, user_details)
     return to_configuration_object(raw_config)
 
 
diff --git a/lib/galaxy/util/config_templates.py b/lib/galaxy/util/config_templates.py
new file mode 100644
index 000000000000..9af06a907fe9
--- /dev/null
+++ b/lib/galaxy/util/config_templates.py
@@ -0,0 +1,167 @@
+"""Utilities for defining user configuration bits from admin templates.
+
+This is capturing code shared by file source templates and object store templates.
+"""
+
+from typing import (
+    Any,
+    Dict,
+    List,
+    Literal,
+    Optional,
+    Protocol,
+    TypeVar,
+    Union,
+)
+
+from boltons.iterutils import remap
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+)
+
+try:
+    from jinja2.nativetypes import NativeEnvironment
+except ImportError:
+    NativeEnvironment = None  # type:ignore[assignment, misc, unused-ignore]
+
+from galaxy.exceptions import (
+    ObjectNotFound,
+    RequestParameterMissingException,
+)
+
+TemplateVariableType = Literal["string", "boolean", "integer"]
+TemplateVariableValueType = Union[str, bool, int]
+TemplateExpansion = str
+MarkdownContent = str
+RawTemplateConfig = Dict[str, Any]
+
+
+class StrictModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class TemplateVariable(StrictModel):
+    name: str
+    help: Optional[MarkdownContent]
+    type: TemplateVariableType
+
+
+class TemplateSecret(StrictModel):
+    name: str
+    help: Optional[MarkdownContent]
+
+
+def expand_raw_config(
+    template_configuration: StrictModel,
+    variables: Dict[str, TemplateVariableValueType],
+    secrets: Dict[str, str],
+    user_details: Dict[str, Any],
+) -> Dict[str, Any]:
+    template_variables = {
+        "variables": variables,
+        "secrets": secrets,
+        "user": user_details,
+    }
+
+    def expand_template(_, key, value):
+        if isinstance(value, str) and "{{" in value and "}}" in value:
+            # NativeEnvironment preserves Python types
+            template = NativeEnvironment().from_string(value)
+            return key, template.render(**template_variables)
+        return key, value
+
+    raw_config = remap(template_configuration.model_dump(), visit=expand_template)
+    return raw_config
+
+
+def verify_vault_configured_if_uses_secrets(catalog, vault_configured: bool, exception_message: str) -> None:
+    if _catalog_uses_secrets(catalog) and not vault_configured:
+        raise Exception(exception_message)
+
+
+def _catalog_uses_secrets(catalog) -> bool:
+    templates = catalog.root
+    for template in templates:
+        if template.secrets and len(template.secrets) > 0:
+            return True
+    return False
+
+
+# cwl-like - convert simple dictionary to list of dictionaries for quickly
+# configuring variables and secrets
+def apply_syntactic_sugar(raw_templates: List[RawTemplateConfig]) -> List[RawTemplateConfig]:
+    templates = []
+    for template in raw_templates:
+        _force_key_to_list(template, "variables")
+        _force_key_to_list(template, "secrets")
+        templates.append(template)
+    return templates
+
+
+def _force_key_to_list(template: RawTemplateConfig, key: str) -> None:
+    value = template.get(key, None)
+    if isinstance(value, dict):
+        value_as_list = []
+        for key_name, key_value in value.items():
+            key_value["name"] = key_name
+            value_as_list.append(key_value)
+        template[key] = value_as_list
+
+
+class TemplateReference(Protocol):
+    template_id: str
+    template_version: int
+
+
+class InstanceDefinition(TemplateReference, Protocol):
+    variables: Dict[str, Any]
+    secrets: Dict[str, str]
+
+
+class Template(Protocol):
+    @property
+    def id(self) -> str:
+        pass
+
+    @property
+    def version(self) -> int:
+        pass
+
+    @property
+    def variables(self) -> Optional[List[TemplateVariable]]:
+        pass
+
+    @property
+    def secrets(self) -> Optional[List[TemplateSecret]]:
+        pass
+
+
+T = TypeVar("T", bound=Template, covariant=True)
+
+
+def find_template(templates: List[T], instance_reference: TemplateReference, what: str) -> T:
+    template_id = instance_reference.template_id
+    template_version = instance_reference.template_version
+    return find_template_by(templates, template_id, template_version, what)
+
+
+def find_template_by(templates: List[T], template_id: str, template_version: int, what: str) -> T:
+    for template in templates:
+        if template.id == template_id and template.version == template_version:
+            return template
+
+    raise ObjectNotFound(f"Could not find a {what} template with id {template_id} and version {template_version}")
+
+
+def validate_secrets_and_variables(instance: InstanceDefinition, template: Template) -> None:
+    secrets = instance.secrets
+    for template_secret in template.secrets or []:
+        name = template_secret.name
+        if name not in secrets:
+            raise RequestParameterMissingException(f"Must define secret '{name}'")
+    variables = instance.variables
+    for template_variable in template.variables or []:
+        name = template_variable.name
+        if name not in variables:
+            raise RequestParameterMissingException(f"Must define variable '{name}'")
diff --git a/lib/galaxy/webapps/galaxy/api/file_sources.py b/lib/galaxy/webapps/galaxy/api/file_sources.py
new file mode 100644
index 000000000000..9e14f9209ea0
--- /dev/null
+++ b/lib/galaxy/webapps/galaxy/api/file_sources.py
@@ -0,0 +1,95 @@
+import logging
+from typing import List
+
+from fastapi import (
+    Body,
+    Path,
+)
+
+from galaxy.files.templates import FileSourceTemplateSummaries
+from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.file_source_instances import (
+    CreateInstancePayload,
+    FileSourceInstancesManager,
+    ModifyInstancePayload,
+    UserFileSourceModel,
+)
+from . import (
+    depends,
+    DependsOnTrans,
+    Router,
+)
+
+log = logging.getLogger(__name__)
+
+router = Router(tags=["file_sources"])
+
+
+UserFileSourceIdPathParam: str = Path(
+    ..., title="User File Source ID", description="The index for a persisted UserFileSourceStore object."
+)
+
+
+@router.cbv
+class FastAPIFileSources:
+    file_source_instances_manager: FileSourceInstancesManager = depends(FileSourceInstancesManager)
+
+    @router.get(
+        "/api/file_source_templates",
+        summary="Get a list of file source templates available to build user defined file sources from",
+        response_description="A list of the configured file source templates.",
+        operation_id="file_sources__templates_index",
+    )
+    def index_templates(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+    ) -> FileSourceTemplateSummaries:
+        return self.file_source_instances_manager.summaries
+
+    @router.post(
+        "/api/file_source_instances",
+        summary="Create a user-bound object store.",
+        operation_id="file_sources__create_instance",
+    )
+    def create(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+        payload: CreateInstancePayload = Body(...),
+    ) -> UserFileSourceModel:
+        return self.file_source_instances_manager.create_instance(trans, payload)
+
+    @router.get(
+        "/api/file_source_instances",
+        summary="Get a list of persisted file source instances defined by the requesting user.",
+        operation_id="file_sources__instances_index",
+    )
+    def instance_index(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+    ) -> List[UserFileSourceModel]:
+        return self.file_source_instances_manager.index(trans)
+
+    @router.get(
+        "/api/file_source_instances/{user_file_source_id}",
+        summary="Get a list of persisted file source instances defined by the requesting user.",
+        operation_id="file_sources__instances_get",
+    )
+    def instances_show(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+        user_file_source_id: str = UserFileSourceIdPathParam,
+    ) -> UserFileSourceModel:
+        return self.file_source_instances_manager.show(trans, user_file_source_id)
+
+    @router.put(
+        "/api/file_source_instances/{user_file_source_id}",
+        summary="Update or upgrade user file source instance.",
+        operation_id="file_sources__instances_update",
+    )
+    def update_instance(
+        self,
+        trans: ProvidesUserContext = DependsOnTrans,
+        user_file_source_id: str = UserFileSourceIdPathParam,
+        payload: ModifyInstancePayload = Body(...),
+    ) -> UserFileSourceModel:
+        return self.file_source_instances_manager.modify_instance(trans, user_file_source_id, payload)
diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py
index 0fdcef4b59cf..da31b21fac5e 100644
--- a/lib/galaxy/webapps/galaxy/buildapp.py
+++ b/lib/galaxy/webapps/galaxy/buildapp.py
@@ -227,6 +227,11 @@ def app_pair(global_conf, load_app_kwds=None, wsgi_preflight=True, **kwargs):
     webapp.add_client_route("/object_store_instances/{user_object_store_id}/edit")
     webapp.add_client_route("/object_store_instances/{user_object_store_id}/upgrade")
     webapp.add_client_route("/object_store_templates/{template_id}/new")
+    webapp.add_client_route("/file_source_instances/create")
+    webapp.add_client_route("/file_source_instances/index")
+    webapp.add_client_route("/file_source_instances/{user_file_source_id}/edit")
+    webapp.add_client_route("/file_source_instances/{user_file_source_id}/upgrade")
+    webapp.add_client_route("/file_source_templates/{template_id}/new")
     webapp.add_client_route("/welcome/new")
     webapp.add_client_route("/visualizations")
     webapp.add_client_route("/visualizations/edit")
diff --git a/packages/util/setup.cfg b/packages/util/setup.cfg
index 70add5d8311c..a3a9d4619367 100644
--- a/packages/util/setup.cfg
+++ b/packages/util/setup.cfg
@@ -53,6 +53,8 @@ jstree =
 template =
     Cheetah3
     future>=1.0.0
+config_template =
+    Jinja2
 
 [options.packages.find]
 exclude =
diff --git a/test/unit/app/managers/test_user_file_sources.py b/test/unit/app/managers/test_user_file_sources.py
new file mode 100644
index 000000000000..96fa43cf5029
--- /dev/null
+++ b/test/unit/app/managers/test_user_file_sources.py
@@ -0,0 +1,109 @@
+from typing import (
+    List,
+    Optional,
+)
+
+from galaxy.files.templates import ConfiguredFileSourceTemplates
+from galaxy.managers.file_source_instances import (
+    CreateInstancePayload,
+    FileSourceInstancesManager,
+    USER_FILE_SOURCES_SCHEME,
+    UserDefinedFileSourcesConfig,
+    UserDefinedFileSourcesImpl,
+    UserFileSourceModel,
+)
+from galaxy.model import UserFileSource
+from galaxy.util.config_templates import RawTemplateConfig
+from .base import BaseTestCase
+
+SIMPLE_FILE_SOURCE_NAME = "myfilesource"
+SIMPLE_FILE_SOURCE_DESCRIPTION = "a description of my file source"
+
+
+class Config:
+    file_source_templates: Optional[List[RawTemplateConfig]] = None
+    file_source_templates_config_file: Optional[str] = None
+
+    def __init__(self, templates: List[RawTemplateConfig]):
+        self.file_source_templates = templates
+
+
+def home_directory_template(tmp_path):
+    return {
+        "id": "home_directory",
+        "name": "Home Directory",
+        "description": "Your Home Directory on this System",
+        "configuration": {
+            "type": "posix",
+            "root": str(tmp_path / "{{ user.username }}"),
+        },
+    }
+
+
+class TestFileSourcesTestCase(BaseTestCase):
+    manager: FileSourceInstancesManager
+    file_sources: UserDefinedFileSourcesImpl
+
+    def test_create_posix(self, tmp_path):
+        self._init_managers(tmp_path)
+        user_file_source = self._create_user_file_source()
+        uri_root = user_file_source.uri_root
+        match = self.file_sources.find_best_match(uri_root)
+        assert match
+        assert match.score == len(uri_root)
+        file_source = match.file_source
+        assert file_source
+        assert file_source.label == SIMPLE_FILE_SOURCE_NAME
+        assert file_source.doc == SIMPLE_FILE_SOURCE_DESCRIPTION
+        assert file_source.get_scheme() == USER_FILE_SOURCES_SCHEME
+        assert file_source.get_uri_root() == uri_root
+
+    def test_io(self, tmp_path):
+        self._init_managers(tmp_path)
+        user_file_source = self._create_user_file_source()
+        uri_root = user_file_source.uri_root
+        match = self.file_sources.find_best_match(uri_root)
+        assert match
+        file_source = match.file_source
+
+        temp_file = tmp_path / "tmp_file"
+        temp_file.write_text("Moo Cow", "utf-8")
+        file_source.write_from("/moo", str(temp_file))
+        target = tmp_path / "round_trip"
+        file_source.realize_to("/moo", target)
+        assert target.read_text("utf-8") == "Moo Cow"
+
+    def test_show(self, tmp_path):
+        self._init_managers(tmp_path)
+        user_file_source = self._create_user_file_source()
+        user_file_source_showed = self.manager.show(self.trans, user_file_source.uuid)
+        assert user_file_source_showed
+        assert user_file_source_showed.uuid == user_file_source.uuid
+        assert user_file_source_showed.id == user_file_source.id
+
+    def _create_user_file_source(self) -> UserFileSourceModel:
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="home_directory",
+            template_version=0,
+            variables={},
+            secrets={},
+        )
+        user_file_source = self.manager.create_instance(self.trans, create_payload)
+        return user_file_source
+
+    def _init_managers(self, tmp_path):
+        file_sources_config = UserDefinedFileSourcesConfig(user_object_store_index_by="uuid")
+        self.app[UserDefinedFileSourcesConfig] = file_sources_config
+        self.app.setup_test_vault()
+        templates_config = Config([home_directory_template(tmp_path)])
+        templates = ConfiguredFileSourceTemplates.from_app_config(
+            templates_config,
+            vault_configured=True,
+        )
+        self.app[ConfiguredFileSourceTemplates] = templates
+        file_sources = self.app[UserDefinedFileSourcesImpl]
+        manager = self.app[FileSourceInstancesManager]
+        self.file_sources = file_sources
+        self.manager = manager
diff --git a/test/unit/files/_util.py b/test/unit/files/_util.py
index 5da87676b120..8ffadef33e9a 100644
--- a/test/unit/files/_util.py
+++ b/test/unit/files/_util.py
@@ -5,6 +5,7 @@
 
 from galaxy.files import (
     ConfiguredFileSources,
+    ConfiguredFileSourcesConf,
     DictFileSourcesUserContext,
 )
 from galaxy.files.plugins import FileSourcePluginsConfig
@@ -127,7 +128,7 @@ def write_from(file_sources, uri, content, user_context=None):
 
 def configured_file_sources(conf_file):
     file_sources_config = FileSourcePluginsConfig()
-    return ConfiguredFileSources(file_sources_config, conf_file=conf_file)
+    return ConfiguredFileSources(file_sources_config, ConfiguredFileSourcesConf(conf_file=conf_file))
 
 
 def assert_simple_file_realize(conf_file, recursive=False, filename="a", contents="a\n", contains=False):
diff --git a/test/unit/files/test_posix.py b/test/unit/files/test_posix.py
index 579b8ee1ca60..adfaa43f78bf 100644
--- a/test/unit/files/test_posix.py
+++ b/test/unit/files/test_posix.py
@@ -12,7 +12,10 @@
     ItemAccessibilityException,
     RequestParameterInvalidException,
 )
-from galaxy.files import ConfiguredFileSources
+from galaxy.files import (
+    ConfiguredFileSources,
+    ConfiguredFileSourcesConf,
+)
 from galaxy.files.plugins import FileSourcePluginsConfig
 from galaxy.files.unittest_utils import (
     setup_root,
@@ -159,7 +162,7 @@ def test_user_ftp_explicit_config():
         "type": "gxftp",
     }
     tmp, root = setup_root()
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[plugin])
+    file_sources = ConfiguredFileSources(file_sources_config, ConfiguredFileSourcesConf(conf_dict=[plugin]))
     user_context = user_context_fixture(user_ftp_dir=root)
     write_file_fixtures(tmp, root)
 
@@ -181,7 +184,9 @@ def test_user_ftp_implicit_config():
         ftp_upload_dir=root,
         ftp_upload_purge=False,
     )
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[], load_stock_plugins=True)
+    file_sources = ConfiguredFileSources(
+        file_sources_config, ConfiguredFileSourcesConf(conf_dict=[]), load_stock_plugins=True
+    )
     user_context = user_context_fixture(user_ftp_dir=root)
     write_file_fixtures(tmp, root)
     assert os.path.exists(os.path.join(root, "a"))
@@ -199,7 +204,9 @@ def test_user_ftp_respects_upload_purge_off():
         ftp_upload_dir=root,
         ftp_upload_purge=True,
     )
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[], load_stock_plugins=True)
+    file_sources = ConfiguredFileSources(
+        file_sources_config, ConfiguredFileSourcesConf(conf_dict=[]), load_stock_plugins=True
+    )
     user_context = user_context_fixture(user_ftp_dir=root)
     write_file_fixtures(tmp, root)
     assert_realizes_as(file_sources, "gxftp://a", "a\n", user_context=user_context)
@@ -211,7 +218,9 @@ def test_user_ftp_respects_upload_purge_on_by_default():
     file_sources_config = FileSourcePluginsConfig(
         ftp_upload_dir=root,
     )
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[], load_stock_plugins=True)
+    file_sources = ConfiguredFileSources(
+        file_sources_config, ConfiguredFileSourcesConf(conf_dict=[]), load_stock_plugins=True
+    )
     user_context = user_context_fixture(user_ftp_dir=root)
     write_file_fixtures(tmp, root)
     assert_realizes_as(file_sources, "gxftp://a", "a\n", user_context=user_context)
@@ -226,7 +235,7 @@ def test_import_dir_explicit_config():
     plugin = {
         "type": "gximport",
     }
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[plugin])
+    file_sources = ConfiguredFileSources(file_sources_config, ConfiguredFileSourcesConf(conf_dict=[plugin]))
     write_file_fixtures(tmp, root)
 
     assert_realizes_as(file_sources, "gximport://a", "a\n")
@@ -237,7 +246,9 @@ def test_import_dir_implicit_config():
     file_sources_config = FileSourcePluginsConfig(
         library_import_dir=root,
     )
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[], load_stock_plugins=True)
+    file_sources = ConfiguredFileSources(
+        file_sources_config, ConfiguredFileSourcesConf(conf_dict=[]), load_stock_plugins=True
+    )
     write_file_fixtures(tmp, root)
 
     assert_realizes_as(file_sources, "gximport://a", "a\n")
@@ -248,7 +259,9 @@ def test_user_import_dir_implicit_config():
     file_sources_config = FileSourcePluginsConfig(
         user_library_import_dir=root,
     )
-    file_sources = ConfiguredFileSources(file_sources_config, conf_dict=[], load_stock_plugins=True)
+    file_sources = ConfiguredFileSources(
+        file_sources_config, ConfiguredFileSourcesConf(conf_dict=[]), load_stock_plugins=True
+    )
 
     write_file_fixtures(tmp, os.path.join(root, EMAIL))
 
diff --git a/test/unit/files/test_template_manager.py b/test/unit/files/test_template_manager.py
new file mode 100644
index 000000000000..5c89405b0c8b
--- /dev/null
+++ b/test/unit/files/test_template_manager.py
@@ -0,0 +1,58 @@
+from galaxy.files.templates import ConfiguredFileSourceTemplates
+from .test_template_models import (
+    LIBRARY_AWS,
+    LIBRARY_HOME_DIRECTORY,
+)
+
+
+class MockConfig:
+    def __init__(self, config_path):
+        self.file_source_templates = None
+        self.file_source_templates_config_file = config_path
+
+
+def test_manager(tmpdir):
+    config_path = tmpdir / "conf.yml"
+    config_path.write_text(LIBRARY_HOME_DIRECTORY, "utf-8")
+    config = MockConfig(config_path)
+    templates = ConfiguredFileSourceTemplates.from_app_config(config)
+    summaries = templates.summaries
+    assert summaries
+    assert len(summaries.root) == 1
+
+
+def test_manager_throws_exception_if_vault_is_required_but_configured(tmpdir):
+    config_path = tmpdir / "conf.yml"
+    config_path.write_text(LIBRARY_AWS, "utf-8")
+    config = MockConfig(config_path)
+    exc = None
+    try:
+        ConfiguredFileSourceTemplates.from_app_config(config, vault_configured=False)
+    except Exception as e:
+        exc = e
+    assert exc, "catalog creation should result in an exception"
+    assert "vault must be configured" in str(exc)
+
+
+def test_manager_with_secrets_is_fine_if_vault_is_required_and_configured(tmpdir):
+    config_path = tmpdir / "conf.yml"
+    config_path.write_text(LIBRARY_AWS, "utf-8")
+    config = MockConfig(config_path)
+    exc = None
+    try:
+        ConfiguredFileSourceTemplates.from_app_config(config, vault_configured=True)
+    except Exception as e:
+        exc = e
+    assert exc is None
+
+
+def test_manager_does_not_throw_exception_if_vault_is_not_required(tmpdir):
+    config_path = tmpdir / "conf.yml"
+    config_path.write_text(LIBRARY_HOME_DIRECTORY, "utf-8")
+    config = MockConfig(config_path)
+    exc = None
+    try:
+        ConfiguredFileSourceTemplates.from_app_config(config, vault_configured=False)
+    except Exception as e:
+        exc = e
+    assert exc is None
diff --git a/test/unit/files/test_template_models.py b/test/unit/files/test_template_models.py
new file mode 100644
index 000000000000..eccfaf073e42
--- /dev/null
+++ b/test/unit/files/test_template_models.py
@@ -0,0 +1,91 @@
+from yaml import safe_load
+
+from galaxy.files.templates.manager import raw_config_to_catalog
+from galaxy.files.templates.models import (
+    FileSourceTemplate,
+    FileSourceTemplateCatalog,
+    PosixFileSourceConfiguration,
+    S3FSFileSourceConfiguration,
+    template_to_configuration,
+)
+
+LIBRARY_AWS = """
+- id: aws_bucket
+  name: Amazon Bucket
+  description: An Amazon S3 Bucket
+  variables:
+    bucket_name:
+      type: string
+      help: Name of bucket to use when connecting to AWS resources.
+  secrets:
+    access_key:
+      help: AWS access key to use when connecting to AWS resources.
+    secret_key:
+      help: AWS secret key to use when connecting to AWS resources.
+  configuration:
+    type: s3fs
+    key: '{{ secrets.access_key}}'
+    secret: '{{ secrets.secret_key}}'
+    bucket: '{{ variables.bucket_name}}'
+"""
+
+
+def test_aws_s3_config():
+    template_library = _parse_template_library(LIBRARY_AWS)
+    s3_template = _assert_has_one_template(template_library)
+    assert s3_template.description == "An Amazon S3 Bucket"
+    configuration_obj = template_to_configuration(
+        s3_template,
+        {"bucket_name": "sec3"},
+        {"access_key": "sec1", "secret_key": "sec2"},
+        user_details={},
+    )
+
+    # expanded configuration should validate with template expansions...
+    assert isinstance(configuration_obj, S3FSFileSourceConfiguration)
+    configuration = configuration_obj.model_dump()
+
+    assert configuration["type"] == "s3fs"
+    assert configuration["key"] == "sec1"
+    assert configuration["secret"] == "sec2"
+    assert configuration["bucket"] == "sec3"
+
+
+LIBRARY_HOME_DIRECTORY = """
+- id: home_directory
+  name: Home Directory
+  description: Your Home Directory on this System
+  configuration:
+    type: posix
+    root: "/home/{{ user.username}}/"
+"""
+
+
+def test_a_posix_template():
+    template_library = _parse_template_library(LIBRARY_HOME_DIRECTORY)
+    posix_template = _assert_has_one_template(template_library)
+    assert posix_template.description == "Your Home Directory on this System"
+    configuration_obj = template_to_configuration(
+        posix_template,
+        {},
+        {},
+        user_details={"username": "foobar"},
+    )
+
+    # expanded configuration should validate with template expansions...
+    assert isinstance(configuration_obj, PosixFileSourceConfiguration)
+    configuration = configuration_obj.model_dump()
+
+    assert configuration["type"] == "posix"
+    assert configuration["root"] == "/home/foobar/"
+
+
+def _assert_has_one_template(catalog: FileSourceTemplateCatalog) -> FileSourceTemplate:
+    assert len(catalog.root) == 1
+    template = catalog.root[0]
+    return template
+
+
+def _parse_template_library(contents: str) -> FileSourceTemplateCatalog:
+    raw_contents = safe_load(contents)
+    return raw_config_to_catalog(raw_contents)