diff --git a/client/src/api/schema/schema.ts b/client/src/api/schema/schema.ts
index 389a25f52387..80f626297b41 100644
--- a/client/src/api/schema/schema.ts
+++ b/client/src/api/schema/schema.ts
@@ -332,6 +332,10 @@ export interface paths {
         /** Get a list of file source templates available to build user defined file sources from */
         get: operations["file_sources__templates_index"];
     };
+    "/api/file_source_templates/{template_id}/{template_version}/oauth2": {
+        /** Template Oauth2 */
+        get: operations["file_sources__template_oauth2"];
+    };
     "/api/folders/{folder_id}/contents": {
         /**
          * Returns a list of a folder's contents (files and sub-folders) with additional metadata about the folder.
@@ -2053,6 +2057,10 @@ export interface paths {
         /** Service Info */
         get: operations["service_info_ga4gh_drs_v1_service_info_get"];
     };
+    "/oauth2_callback": {
+        /** Callback entry point for remote resource responses with OAuth2 authorization codes */
+        get: operations["oauth2_callback_oauth2_callback_get"];
+    };
 }
 
 export type webhooks = Record<string, never>;
@@ -3151,6 +3159,8 @@ export interface components {
             template_id: string;
             /** Template Version */
             template_version: number;
+            /** Uuid */
+            uuid?: string | null;
             /** Variables */
             variables: {
                 [key: string]: (string | boolean | number) | undefined;
@@ -5178,7 +5188,7 @@ export interface components {
              * Type
              * @enum {string}
              */
-            type: "ftp" | "posix" | "s3fs" | "azure";
+            type: "ftp" | "posix" | "s3fs" | "azure" | "dropbox" | "googledrive";
             /** Variables */
             variables?:
                 | (
@@ -9954,6 +9964,11 @@ export interface components {
              */
             updated_count: number;
         };
+        /** OAuth2Info */
+        OAuth2Info: {
+            /** Authorize Url */
+            authorize_url: string;
+        };
         /** ObjectExportTaskResponse */
         ObjectExportTaskResponse: {
             /**
@@ -10479,6 +10494,7 @@ export interface components {
         /** PluginStatus */
         PluginStatus: {
             connection?: components["schemas"]["PluginAspectStatus"] | null;
+            oauth2_access_token_generation?: components["schemas"]["PluginAspectStatus"] | null;
             template_definition: components["schemas"]["PluginAspectStatus"];
             template_settings?: components["schemas"]["PluginAspectStatus"] | null;
         };
@@ -12790,7 +12806,7 @@ export interface components {
              * Type
              * @enum {string}
              */
-            type: "ftp" | "posix" | "s3fs" | "azure";
+            type: "ftp" | "posix" | "s3fs" | "azure" | "dropbox" | "googledrive";
             /** Uri Root */
             uri_root: string;
             /** Uuid */
@@ -15178,6 +15194,35 @@ export interface operations {
             };
         };
     };
+    file_sources__template_oauth2: {
+        /** Template Oauth2 */
+        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 template ID of the target file source template. */
+            /** @description The template version of the target file source template. */
+            path: {
+                template_id: string;
+                template_version: number;
+            };
+        };
+        responses: {
+            /** @description OAuth2 authorization url to redirect user to prior to creation. */
+            200: {
+                content: {
+                    "application/json": components["schemas"]["OAuth2Info"];
+                };
+            };
+            /** @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.
@@ -25626,4 +25671,32 @@ export interface operations {
             };
         };
     };
+    oauth2_callback_oauth2_callback_get: {
+        /** Callback entry point for remote resource responses with OAuth2 authorization codes */
+        parameters: {
+            /** @description Base-64 encoded JSON used to route request within Galaxy. */
+            query: {
+                state: string;
+                code: string;
+            };
+            /** @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": Record<string, never>;
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
 }
diff --git a/client/src/components/ConfigTemplates/ConfigurationTestSummary.vue b/client/src/components/ConfigTemplates/ConfigurationTestSummary.vue
index be17f1593821..b781c6eae665 100644
--- a/client/src/components/ConfigTemplates/ConfigurationTestSummary.vue
+++ b/client/src/components/ConfigTemplates/ConfigurationTestSummary.vue
@@ -15,6 +15,9 @@ defineProps<Props>();
 <template>
     <BListGroup v-if="testResults">
         <ConfigurationTestItem :status="testResults?.template_definition" />
+        <ConfigurationTestItem
+            v-if="testResults?.oauth2_access_token_generation != null"
+            :status="testResults?.oauth2_access_token_generation" />
         <ConfigurationTestItem :status="testResults?.template_settings" />
         <ConfigurationTestItem :status="testResults?.connection" />
     </BListGroup>
diff --git a/client/src/components/ConfigTemplates/useConfigurationTesting.ts b/client/src/components/ConfigTemplates/useConfigurationTesting.ts
index 9c2de8a60c7f..c9b803ae5ef7 100644
--- a/client/src/components/ConfigTemplates/useConfigurationTesting.ts
+++ b/client/src/components/ConfigTemplates/useConfigurationTesting.ts
@@ -43,6 +43,7 @@ export function useConfigurationTesting() {
 export function useConfigurationTemplateCreation<T extends TemplateSummary, R>(
     what: string,
     template: Ref<T>,
+    uuid: Ref<string | undefined>,
     test: (payload: CreateInstancePayload) => Promise<{ data: PluginStatus }>,
     create: (payload: CreateInstancePayload) => Promise<{ data: R }>,
     onCreate: (result: R) => unknown
@@ -52,6 +53,9 @@ export function useConfigurationTemplateCreation<T extends TemplateSummary, R>(
 
     async function onSubmit(formData: any) {
         const payload = createFormDataToPayload(template.value, formData);
+        if (uuid.value) {
+            payload.uuid = uuid.value;
+        }
         let pluginStatus;
         try {
             testRunning.value = true;
diff --git a/client/src/components/FileSources/Instances/CreateForm.vue b/client/src/components/FileSources/Instances/CreateForm.vue
index dc92fe118bd9..eb11814864e0 100644
--- a/client/src/components/FileSources/Instances/CreateForm.vue
+++ b/client/src/components/FileSources/Instances/CreateForm.vue
@@ -8,6 +8,7 @@ import { create, test } from "./services";
 
 interface CreateFormProps {
     template: FileSourceTemplateSummary;
+    uuid?: string;
 }
 const props = defineProps<CreateFormProps>();
 const title = "Create a new file source for your data";
@@ -20,6 +21,7 @@ const { ActionSummary, error, inputs, InstanceForm, onSubmit, submitTitle, loadi
     useConfigurationTemplateCreation(
         "file source",
         toRef(props, "template"),
+        toRef(props, "uuid"),
         test,
         create,
         (fileSource: UserFileSourceModel) => emit("created", fileSource)
diff --git a/client/src/components/FileSources/Instances/CreateInstance.vue b/client/src/components/FileSources/Instances/CreateInstance.vue
index 9b93e2d7c269..cee1849cc30a 100644
--- a/client/src/components/FileSources/Instances/CreateInstance.vue
+++ b/client/src/components/FileSources/Instances/CreateInstance.vue
@@ -1,17 +1,22 @@
 <script setup lang="ts">
-import { computed } from "vue";
+import { computed, watch } from "vue";
 
 import type { UserFileSourceModel } from "@/api/fileSources";
 import { useFileSourceTemplatesStore } from "@/stores/fileSourceTemplatesStore";
 
 import { useInstanceRouting } from "./routing";
+import { getOAuth2Info } from "./services";
 
 import CreateForm from "@/components/FileSources/Instances/CreateForm.vue";
 import LoadingSpan from "@/components/LoadingSpan.vue";
 
 interface Props {
     templateId: string;
+    uuid?: string;
 }
+
+const OAUTH2_TYPES = ["dropbox", "googledrive"];
+
 const fileSourceTemplatesStore = useFileSourceTemplatesStore();
 fileSourceTemplatesStore.fetchTemplates();
 
@@ -19,16 +24,40 @@ const { goToIndex } = useInstanceRouting();
 
 const props = defineProps<Props>();
 const template = computed(() => fileSourceTemplatesStore.getLatestTemplate(props.templateId));
+const requiresOAuth2AuthorizeRedirect = computed(() => {
+    const templateValue = template.value;
+    return props.uuid == undefined && templateValue && OAUTH2_TYPES.indexOf(templateValue.type) >= 0;
+});
 
 async function onCreated(objectStore: UserFileSourceModel) {
     const message = `Created file source ${objectStore.name}`;
     goToIndex({ message });
 }
+
+watch(
+    requiresOAuth2AuthorizeRedirect,
+    async (requiresAuth) => {
+        const templateValue = template.value;
+        if (templateValue && requiresAuth) {
+            const { data } = await getOAuth2Info({
+                template_id: templateValue.id,
+                template_version: templateValue.version || 0,
+            });
+            window.location.href = data.authorize_url;
+        } else {
+            console.log("skipping this...");
+        }
+    },
+    { immediate: true }
+);
 </script>
 
 <template>
     <div>
         <LoadingSpan v-if="!template" message="Loading file source templates" />
-        <CreateForm v-else :template="template" @created="onCreated"></CreateForm>
+        <LoadingSpan
+            v-else-if="requiresOAuth2AuthorizeRedirect"
+            message="Fetching redirect information, you will need to authorize Galaxy to have access to this resource remotely" />
+        <CreateForm v-else :uuid="uuid" :template="template" @created="onCreated"></CreateForm>
     </div>
 </template>
diff --git a/client/src/components/FileSources/Instances/services.ts b/client/src/components/FileSources/Instances/services.ts
index 7e9ff219d708..04a24a12a4f3 100644
--- a/client/src/components/FileSources/Instances/services.ts
+++ b/client/src/components/FileSources/Instances/services.ts
@@ -9,6 +9,10 @@ export const testInstance = fetcher
     .create();
 export const update = fetcher.path("/api/file_source_instances/{user_file_source_id}").method("put").create();
 export const testUpdate = fetcher.path("/api/file_source_instances/{user_file_source_id}/test").method("post").create();
+export const getOAuth2Info = fetcher
+    .path("/api/file_source_templates/{template_id}/{template_version}/oauth2")
+    .method("get")
+    .create();
 
 export async function hide(instance: UserFileSourceModel) {
     const payload = { hidden: true };
diff --git a/client/src/components/ObjectStore/Instances/CreateForm.vue b/client/src/components/ObjectStore/Instances/CreateForm.vue
index 3eff5eedf286..85d4d62f51d3 100644
--- a/client/src/components/ObjectStore/Instances/CreateForm.vue
+++ b/client/src/components/ObjectStore/Instances/CreateForm.vue
@@ -9,6 +9,7 @@ import { create, test } from "./services";
 
 interface CreateFormProps {
     template: ObjectStoreTemplateSummary;
+    uuid?: string;
 }
 const props = defineProps<CreateFormProps>();
 const title = "Create a new storage location for your data";
@@ -21,6 +22,7 @@ const { ActionSummary, error, inputs, InstanceForm, onSubmit, submitTitle, loadi
     useConfigurationTemplateCreation(
         "storage location",
         toRef(props, "template"),
+        toRef(props, "uuid"),
         test,
         create,
         (instance: UserConcreteObjectStore) => emit("created", instance)
diff --git a/client/src/entry/analysis/router.js b/client/src/entry/analysis/router.js
index 811544520bd9..d1b560e70c28 100644
--- a/client/src/entry/analysis/router.js
+++ b/client/src/entry/analysis/router.js
@@ -408,7 +408,10 @@ export function getRouter(Galaxy) {
                     {
                         path: "file_source_templates/:templateId/new",
                         component: CreateFileSourceInstance,
-                        props: true,
+                        props: (route) => ({
+                            templateId: route.params.templateId,
+                            uuid: route.query.uuid,
+                        }),
                     },
                     {
                         path: "pages/create",
diff --git a/doc/source/admin/data.md b/doc/source/admin/data.md
index 1e9a3ea90952..2f9338ab5ff5 100644
--- a/doc/source/admin/data.md
+++ b/doc/source/admin/data.md
@@ -350,6 +350,19 @@ configuration).
 
 ![](file_source_azure_configuration.png)
 
+#### ``dropbox``
+
+The syntax for the ``configuration`` section of ``dropbox`` templates looks like this.
+
+![](file_source_dropbox_configuration_template.png)
+
+At runtime, after the ``configuration`` template is expanded, the resulting dictionary
+passed to Galaxy's file source plugin infrastructure looks like this and should match a subset
+of what you'd be able to add directly to ``file_sources_conf.yml`` (Galaxy's global file source
+configuration).
+
+![](file_source_dropbox_configuration.png)
+
 ### YAML Syntax
 
 ![galaxy.files.templates.models](file_source_templates.png)
@@ -398,6 +411,74 @@ and you are comfortable with it storing your user's secrets.
 :language: yaml
 ```
 
+### Production OAuth 2.0 File Source Templates
+
+Unlike the examples in the previous section. These examples will require a bit of
+configuration on the part of the admin. This is to obtain client credentials from the
+external service and register an OAuth 2.0 redirection callback with the remote service.
+
+#### Dropbox
+
+Once you have OAuth 2.0 client credentials from Dropbox (called ``oauth2_client_id`` and
+``oauth2_client_secret`` here), the following configuration can be used configure your Galaxy
+instance to enable Dropbox.
+
+```{literalinclude} ../../../lib/galaxy/files/templates/examples/production_dropbox.yml
+:language: yaml
+```
+
+To use this template - you'll need to make your credentials available to Galaxy's
+web and job handler processes using the environment variables ``GALAXY_DROPBOX_APP_CLIENT_ID``
+and ``GALAXY_DROPBOX_APP_CLIENT_SECRET``. Your jobs themselves do not require
+these secrets to be set and will not be given the secrets.
+
+If you'd like to configure these secrets explicit - you can configure them explicitly
+in the configuration. If your configuration file is managed by Ansible, these secrets
+could potentially be populated from your Ansible vault.
+
+```{literalinclude} ../../../lib/galaxy/files/templates/examples/dropbox_client_secrets_explicit.yml
+:language: yaml
+```
+
+To obtain the OAuth 2.0 credentials from Dropbox, you'll need to navigate to your
+[Dropbox Apps](https://www.dropbox.com/developers/apps) and create a new app 
+for your Galaxy instance with the "Create app" button.
+
+![Screenshot of Dropbox](dropbox_create_button.png)
+
+The only option available is "Scoped access" and this works fine for typical
+Galaxy use cases. You will however want to click "Full Dropbox" to request
+full access to your user's account. You will also need to give your "App" a name
+here, this should likely be something related to your Galaxy instances name.
+
+![Screenshot of Dropbox](dropbox_create_app_options_1.png)
+
+After your app is created, you'll be presented with a management screen for it.
+The first thing you'll want to do is navigate to the permissions tab and enable
+permissions to read and write to files and directories so the file source plugin
+works properly:
+
+![Screenshot of Dropbox](dropbox_scopes.png)
+
+Next, navigate back to the "Settings" tab. You'll need to register a callback
+for your Galaxy instance (it will need HTTPS enabled). This should be the URL
+to your Galaxy instance with ``oauth2_callback`` appended to it.
+
+![Screenshot of Dropbox](dropbox_callback.png)
+
+Finally you'll be able to find the ``oauth2_client_id`` and ``oauth2_client_secret``
+to configured your Galaxy with on this settings page.
+
+![Screenshot of Dropbox](dropbox_client_creds.png)
+
+Until you have 50 users, your App will be considered a "development" application.
+The upshot of this is that your user's will get a scary message during authorization
+but there seems to be no way around this. 50 users would definitely be considered
+a production Galaxy instance but Dropbox operates on a different scale.
+
+For more information on what Dropbox considers a "development" app versus a "production"
+app - checkout the [Dropbox documentation](https://www.dropbox.com/developers/reference/developer-guide#production-approval).
+
 ## Playing Nicer with Ansible
 
 Many large instances of Galaxy are configured with Ansible and much of the existing administrator
@@ -604,3 +685,70 @@ variable is not defined at runtime.
 ```{literalinclude} ../../../lib/galaxy/files/templates/examples/admin_secrets_with_defaults.yml
 :language: yaml
 ```
+
+## OAuth 2.0 Enabled Configurations
+
+[OAuth 2.0](https://oauth.net/2/) has become an industry standard for allowing users
+of various services (e.g. Dropbox or Google Drive) to authorize other services (e.g. Galaxy)
+fine grained access to the services. There is a bit of a dance the services need to do
+but the result can be a fairly nice end-user experience. The framework for configuring
+user defined data access templates can support OAuth 2.0.
+
+Galaxy keeps track of which plugin ``type``s (currently only file source types) require
+OAuth2 to work properly and will take care of authorization redirection, saving refresh tokens,
+etc.. implicitly. One such ``type`` is ``dropbox``. Here is the production Dropbox
+template distributed with Galaxy.
+
+```{literalinclude} ../../../lib/galaxy/files/templates/examples/production_dropbox.yml
+:language: yaml
+```
+
+OAuth2 enabled plugin types include template definitions that include ``oauth2_client_id``
+and ``oauth2_client_secret`` in the configuration (as shown in the following specification
+and in the above examples).
+
+![](file_source_dropbox_configuration_template.png)
+
+The above example defines these secrets using environment variables but they can stored in
+Galaxy's Vault explicitly by the admin or written right to the configuration files as shown
+in the next two examples:
+
+```{literalinclude} ../../../lib/galaxy/files/templates/examples/dropbox_client_secrets_in_vault.yml
+:language: yaml
+```
+
+```{literalinclude} ../../../lib/galaxy/files/templates/examples/dropbox_client_secrets_explicit.yml
+:language: yaml
+```
+
+Looking at the configuration objects that get generated at runtime
+from these templates though - ``oauth2_client_id`` and ``oauth2_client_secret`` no longer
+appear and instead have been replaced with a ``oauth2_access_token`` parameter.
+Galaxy will take care of stripping out the client (e.g. Galaxy server) information and
+replacing it with short-term access tokens generated for the user's resources.
+
+![](file_source_dropbox_configuration.png)
+
+Normally, a UUID is created for each user configured instance object and this is used
+to store the template's explicitly listed secrets in Galaxy's Vault. For OAuth 2.0
+plugin types - before user's are even prompted for configuration metadata they are redirected
+to the remote service and prompted to authorize Galaxy to act on their behalf when using
+the remote service. If they authorize this, the remote service will send an [`authorization
+code`](https://oauth.net/2/grant-types/authorization-code/) to ``https://<galaxy_url>/oauth2_callback`` along with state information
+to recover which instance is being configured. At this point, Galaxy will fetch a [`refresh token`](https://oauth.net/2/refresh-tokens/) from the remote resource using the
+supplied authorization code. The refresh token is stored in the Vault in key associated with
+the UUID of the object that will be created when the user finishes the creation process.
+Specifically it is stored at 
+
+```
+/galaxy/user/<user_id>/file_source_config/<file_source_instance_uuid>/_oauth2_refresh_token
+```
+
+Here is the prefix at the end of ``_`` is indicating that Galaxy is managing this instead of
+it being listed explicitly in a ``secrets`` section of the template configuration like the
+explicit Vault secrets discussed in this document.
+
+Galaxy knows how to fetch an [`access token`](https://oauth.net/2/access-tokens/) from this
+refresh token that is actually used to interact with the remote resource. This is the property
+``oauth2_access_token`` that is injected into the configuration object shown above and passed
+along to the actual object store or file source plugin implementation.
diff --git a/doc/source/admin/dropbox_callback.png b/doc/source/admin/dropbox_callback.png
new file mode 100644
index 000000000000..0125e94fec45
Binary files /dev/null and b/doc/source/admin/dropbox_callback.png differ
diff --git a/doc/source/admin/dropbox_client_creds.png b/doc/source/admin/dropbox_client_creds.png
new file mode 100644
index 000000000000..3a029cd3c85a
Binary files /dev/null and b/doc/source/admin/dropbox_client_creds.png differ
diff --git a/doc/source/admin/dropbox_create_app_options_1.png b/doc/source/admin/dropbox_create_app_options_1.png
new file mode 100644
index 000000000000..839060a05783
Binary files /dev/null and b/doc/source/admin/dropbox_create_app_options_1.png differ
diff --git a/doc/source/admin/dropbox_create_button.png b/doc/source/admin/dropbox_create_button.png
new file mode 100644
index 000000000000..6082eb174981
Binary files /dev/null and b/doc/source/admin/dropbox_create_button.png differ
diff --git a/doc/source/admin/dropbox_scopes.png b/doc/source/admin/dropbox_scopes.png
new file mode 100644
index 000000000000..5a2643d0a0ff
Binary files /dev/null and b/doc/source/admin/dropbox_scopes.png differ
diff --git a/doc/source/admin/file_source_dropbox_configuration.png b/doc/source/admin/file_source_dropbox_configuration.png
new file mode 100644
index 000000000000..b658fbd8399c
Binary files /dev/null and b/doc/source/admin/file_source_dropbox_configuration.png differ
diff --git a/doc/source/admin/file_source_dropbox_configuration_template.png b/doc/source/admin/file_source_dropbox_configuration_template.png
new file mode 100644
index 000000000000..98ea13daec76
Binary files /dev/null and b/doc/source/admin/file_source_dropbox_configuration_template.png differ
diff --git a/doc/source/admin/file_source_templates.png b/doc/source/admin/file_source_templates.png
index 5d1556b69269..6b8b2ac9c6cc 100644
Binary files a/doc/source/admin/file_source_templates.png and b/doc/source/admin/file_source_templates.png differ
diff --git a/doc/source/admin/gen_diagrams.py b/doc/source/admin/gen_diagrams.py
index 2627420177eb..145d475e2b83 100644
--- a/doc/source/admin/gen_diagrams.py
+++ b/doc/source/admin/gen_diagrams.py
@@ -8,6 +8,8 @@
 from galaxy.files.templates.models import (
     AzureFileSourceConfiguration,
     AzureFileSourceTemplateConfiguration,
+    DropboxFileSourceConfiguration,
+    DropboxFileSourceTemplateConfiguration,
     FileSourceTemplate,
     FtpFileSourceConfiguration,
     FtpFileSourceTemplateConfiguration,
@@ -47,6 +49,8 @@
     FileSourceTemplate: "file_source_templates",
     AzureFileSourceTemplateConfiguration: "file_source_azure_configuration_template",
     AzureFileSourceConfiguration: "file_source_azure_configuration",
+    DropboxFileSourceTemplateConfiguration: "file_source_dropbox_configuration_template",
+    DropboxFileSourceConfiguration: "file_source_dropbox_configuration",
     PosixFileSourceTemplateConfiguration: "file_source_posix_configuration_template",
     PosixFileSourceConfiguration: "file_source_posix_configuration",
     S3FSFileSourceTemplateConfiguration: "file_source_s3fs_configuration_template",
diff --git a/lib/galaxy/app_unittest_utils/galaxy_mock.py b/lib/galaxy/app_unittest_utils/galaxy_mock.py
index 95715c042543..a7a76ef9a891 100644
--- a/lib/galaxy/app_unittest_utils/galaxy_mock.py
+++ b/lib/galaxy/app_unittest_utils/galaxy_mock.py
@@ -329,7 +329,7 @@ def __init__(self, app=None, user=None, history=None, **kwargs):
         self.security = self.app.security
         self.history = history
 
-        self.request: Any = Bunch(headers={}, is_body_readable=False, host="request.host")
+        self.request: Any = Bunch(headers={}, is_body_readable=False, host="request.host", url_path="mock/url/path")
         self.response: Any = Bunch(headers={}, set_content_type=lambda i: None)
 
     @property
diff --git a/lib/galaxy/files/sources/dropbox.py b/lib/galaxy/files/sources/dropbox.py
index 70c994e3e2e4..e8bad36d2812 100644
--- a/lib/galaxy/files/sources/dropbox.py
+++ b/lib/galaxy/files/sources/dropbox.py
@@ -30,6 +30,8 @@ def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None)
         # accessToken has been renamed to access_token in fs.dropboxfs 1.0
         if "accessToken" in props:
             props["access_token"] = props.pop("accessToken")
+        if "oauth2_access_token" in props:
+            props["access_token"] = props.pop("oauth2_access_token")
 
         try:
             handle = DropboxFS(**{**props, **extra_props})
diff --git a/lib/galaxy/files/sources/googledrive.py b/lib/galaxy/files/sources/googledrive.py
index c5e777d6b78d..e809d4cd03bf 100644
--- a/lib/galaxy/files/sources/googledrive.py
+++ b/lib/galaxy/files/sources/googledrive.py
@@ -17,6 +17,9 @@ class GoogleDriveFilesSource(PyFilesystem2FilesSource):
 
     def _open_fs(self, user_context=None, opts: Optional[FilesSourceOptions] = None):
         props = self._serialization_props(user_context)
+        access_token = props.pop("oauth2_access_token")
+        if access_token:
+            props["token"] = access_token
         credentials = Credentials(**props)
         handle = GoogleDriveFS(credentials)
         return handle
diff --git a/lib/galaxy/files/templates/__init__.py b/lib/galaxy/files/templates/__init__.py
index aa8faf85fd30..2aef4eb906ed 100644
--- a/lib/galaxy/files/templates/__init__.py
+++ b/lib/galaxy/files/templates/__init__.py
@@ -4,6 +4,8 @@
     FileSourceTemplate,
     FileSourceTemplateSummaries,
     FileSourceTemplateType,
+    get_oauth2_config,
+    get_oauth2_config_or_none,
     template_to_configuration,
 )
 
@@ -13,5 +15,7 @@
     "FileSourceTemplate",
     "FileSourceTemplateSummaries",
     "FileSourceTemplateType",
+    "get_oauth2_config",
+    "get_oauth2_config_or_none",
     "template_to_configuration",
 )
diff --git a/lib/galaxy/files/templates/examples/dropbox_client_secrets_explicit.yml b/lib/galaxy/files/templates/examples/dropbox_client_secrets_explicit.yml
new file mode 100644
index 000000000000..4206e9997b62
--- /dev/null
+++ b/lib/galaxy/files/templates/examples/dropbox_client_secrets_explicit.yml
@@ -0,0 +1,7 @@
+- id: dropbox
+  name: Dropbox
+  description: Connect to your Dropbox account to download and upload files.
+  configuration:
+    type: dropbox
+    oauth2_client_id: abcdefgh
+    oauth2_client_secret: ijklmnopqr
diff --git a/lib/galaxy/files/templates/examples/dropbox_client_secrets_in_vault.yml b/lib/galaxy/files/templates/examples/dropbox_client_secrets_in_vault.yml
new file mode 100644
index 000000000000..4ed3be1ec510
--- /dev/null
+++ b/lib/galaxy/files/templates/examples/dropbox_client_secrets_in_vault.yml
@@ -0,0 +1,15 @@
+- id: dropbox
+  name: Dropbox
+  description: Connect to your Dropbox account to download and upload files.
+  configuration:
+    type: dropbox
+    oauth2_client_id: "{{ environment.oauth2_client_id }}"
+    oauth2_client_secret: "{{ environment.oauth2_client_secret }}"
+    writable: true
+  environment:
+    oauth2_client_id:
+      type: secret
+      vault_key: "dropbox_file_source/client_id"
+    oauth2_client_secret:
+      type: secret
+      vault_key: "dropbox_file_source/client_secret"
diff --git a/lib/galaxy/files/templates/examples/production_dropbox.yml b/lib/galaxy/files/templates/examples/production_dropbox.yml
new file mode 100644
index 000000000000..61ac82ad1041
--- /dev/null
+++ b/lib/galaxy/files/templates/examples/production_dropbox.yml
@@ -0,0 +1,15 @@
+- id: dropbox
+  name: Dropbox
+  description: Connect to your Dropbox account to download and upload files.
+  configuration:
+    type: dropbox
+    oauth2_client_id: "{{ environment.oauth2_client_id }}"
+    oauth2_client_secret: "{{ environment.oauth2_client_secret }}"
+    writable: true
+  environment:
+    oauth2_client_id:
+      type: variable
+      variable: GALAXY_DROPBOX_APP_CLIENT_ID
+    oauth2_client_secret:
+      type: variable
+      variable: GALAXY_DROPBOX_APP_CLIENT_SECRET
diff --git a/lib/galaxy/files/templates/examples/production_google_drive.yml b/lib/galaxy/files/templates/examples/production_google_drive.yml
new file mode 100644
index 000000000000..164b162ef386
--- /dev/null
+++ b/lib/galaxy/files/templates/examples/production_google_drive.yml
@@ -0,0 +1,15 @@
+- id: google_drive
+  name: Export to Google Drive
+  description: This file source allows export to your Google Drive but can only read files uploaded by this Galaxy instance.
+  configuration:
+    type: googledrive
+    oauth2_client_id: "{{ environment.oauth2_client_id }}"
+    oauth2_client_secret: "{{ environment.oauth2_client_secret }}"
+    writable: true
+  environment:
+    oauth2_client_id:
+      type: variable
+      variable: GALAXY_GOOGLE_DRIVE_APP_CLIENT_ID
+    oauth2_client_secret:
+      type: variable
+      variable: GALAXY_GOOGLE_DRIVE_APP_CLIENT_SECRET
diff --git a/lib/galaxy/files/templates/models.py b/lib/galaxy/files/templates/models.py
index 2e22cc904c30..463b8d81d308 100644
--- a/lib/galaxy/files/templates/models.py
+++ b/lib/galaxy/files/templates/models.py
@@ -11,9 +11,14 @@
 from pydantic import RootModel
 
 from galaxy.util.config_templates import (
+    ConfiguredOAuth2Sources,
     EnvironmentDict,
     expand_raw_config,
+    get_oauth2_config_from,
+    ImplicitConfigurationParameters,
     MarkdownContent,
+    merge_implicit_parameters,
+    OAuth2Configuration,
     populate_default_variables,
     SecretsDict,
     StrictModel,
@@ -25,7 +30,7 @@
     UserDetailsDict,
 )
 
-FileSourceTemplateType = Literal["ftp", "posix", "s3fs", "azure"]
+FileSourceTemplateType = Literal["ftp", "posix", "s3fs", "azure", "dropbox", "googledrive"]
 
 
 class PosixFileSourceTemplateConfiguration(StrictModel):
@@ -42,6 +47,54 @@ class PosixFileSourceConfiguration(StrictModel):
     writable: bool = False
 
 
+class OAuth2TemplateConfiguration:
+    oauth2_client_id: Union[str, TemplateExpansion]
+    oauth2_client_secret: Union[str, TemplateExpansion]
+
+
+class DropboxFileSourceTemplateConfiguration(OAuth2TemplateConfiguration, StrictModel):
+    type: Literal["dropbox"]
+    writable: Union[bool, TemplateExpansion] = False
+    oauth2_client_id: Union[str, TemplateExpansion]
+    oauth2_client_secret: Union[str, TemplateExpansion]
+    template_start: Optional[str] = None
+    template_end: Optional[str] = None
+
+
+class OAuth2FileSourceConfiguration:
+    oauth2_access_token: str
+
+
+class DropboxFileSourceConfiguration(OAuth2FileSourceConfiguration, StrictModel):
+    type: Literal["dropbox"]
+    writable: bool = False
+    oauth2_access_token: str
+
+
+class GoogleDriveFileSourceTemplateConfiguration(OAuth2TemplateConfiguration, StrictModel):
+    type: Literal["googledrive"]
+    writable: Union[bool, TemplateExpansion] = False
+    oauth2_client_id: Union[str, TemplateExpansion]
+    oauth2_client_secret: Union[str, TemplateExpansion]
+    # Will default to https://www.googleapis.com/auth/drive.file, which provides
+    # access to a folder specific to your Galaxy instance. Ideally we would use
+    # https://www.googleapis.com/auth/drive but that would require becoming
+    # Google verified - https://support.google.com/cloud/answer/13464321#ss-rs-requirements.
+    # That seems like a onerous process and I don't know how it would
+    # work in the context of an open source project like Galaxy, I am
+    # adding the extension point here for the brave individual that would like
+    # to use it but I expect it isn't practical for the typical admin.
+    oauth2_scope: Optional[List[Union[str, TemplateExpansion]]] = None
+    template_start: Optional[str] = None
+    template_end: Optional[str] = None
+
+
+class GoogleDriveFileSourceConfiguration(OAuth2FileSourceConfiguration, StrictModel):
+    type: Literal["googledrive"]
+    writable: bool = False
+    oauth2_access_token: str
+
+
 class S3FSFileSourceTemplateConfiguration(StrictModel):
     type: Literal["s3fs"]
     endpoint_url: Optional[Union[str, TemplateExpansion]] = None
@@ -109,12 +162,16 @@ class AzureFileSourceConfiguration(StrictModel):
     S3FSFileSourceTemplateConfiguration,
     FtpFileSourceTemplateConfiguration,
     AzureFileSourceTemplateConfiguration,
+    DropboxFileSourceTemplateConfiguration,
+    GoogleDriveFileSourceTemplateConfiguration,
 ]
 FileSourceConfiguration = Union[
     PosixFileSourceConfiguration,
     S3FSFileSourceConfiguration,
     FtpFileSourceConfiguration,
     AzureFileSourceConfiguration,
+    DropboxFileSourceConfiguration,
+    GoogleDriveFileSourceConfiguration,
 ]
 
 
@@ -163,10 +220,12 @@ def template_to_configuration(
     secrets: SecretsDict,
     user_details: UserDetailsDict,
     environment: EnvironmentDict,
+    implicit: Optional[ImplicitConfigurationParameters] = None,
 ) -> FileSourceConfiguration:
     configuration_template = template.configuration
     populate_default_variables(template.variables, variables)
     raw_config = expand_raw_config(configuration_template, variables, secrets, user_details, environment)
+    merge_implicit_parameters(raw_config, implicit)
     return to_configuration_object(raw_config)
 
 
@@ -175,9 +234,36 @@ def template_to_configuration(
     "posix": PosixFileSourceConfiguration,
     "s3fs": S3FSFileSourceConfiguration,
     "azure": AzureFileSourceConfiguration,
+    "dropbox": DropboxFileSourceConfiguration,
+    "googledrive": GoogleDriveFileSourceConfiguration,
 }
 
 
+OAUTH2_CONFIGURED_SOURCES: ConfiguredOAuth2Sources = {
+    "dropbox": OAuth2Configuration(
+        authorize_url="https://www.dropbox.com/oauth2/authorize",
+        authorize_params={"token_access_type": "offline"},
+        token_url="https://api.dropbox.com/oauth2/token",
+    ),
+    "googledrive": OAuth2Configuration(
+        authorize_url="https://accounts.google.com/o/oauth2/v2/auth",
+        authorize_params={"access_type": "offline", "prompt": "consent"},
+        token_url="https://oauth2.googleapis.com/token",
+        scope=["https://www.googleapis.com/auth/drive.file"],
+    ),
+}
+
+
+def get_oauth2_config(template: FileSourceTemplate) -> OAuth2Configuration:
+    return get_oauth2_config_from(template, OAUTH2_CONFIGURED_SOURCES)
+
+
+def get_oauth2_config_or_none(template: FileSourceTemplate) -> Optional[OAuth2Configuration]:
+    if template.configuration.type not in OAUTH2_CONFIGURED_SOURCES:
+        return None
+    return get_oauth2_config(template)
+
+
 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.")
diff --git a/lib/galaxy/managers/_config_templates.py b/lib/galaxy/managers/_config_templates.py
index 303ef4d1a015..e504802bb5ae 100644
--- a/lib/galaxy/managers/_config_templates.py
+++ b/lib/galaxy/managers/_config_templates.py
@@ -6,6 +6,7 @@
     Dict,
     List,
     Optional,
+    Type,
     TypeVar,
     Union,
 )
@@ -14,6 +15,7 @@
 from typing_extensions import TypedDict
 
 from galaxy.exceptions import (
+    InconsistentDatabase,
     ObjectNotFound,
     RequestParameterInvalidException,
     RequestParameterMissingException,
@@ -32,9 +34,14 @@
     UserVaultWrapper,
     Vault,
 )
+from galaxy.util import config_templates  # defer get_token_from_refresh_raw access for monkey patching
 from galaxy.util.config_templates import (
     EnvironmentDict,
     find_template_by,
+    ImplicitConfigurationParameters,
+    OAuth2ClientPair,
+    OAuth2Configuration,
+    PluginAspectStatus,
     secrets_as_dict,
     SecretsDict,
     Template,
@@ -46,6 +53,7 @@
     validate_no_extra_variables_defined,
     validate_specified_datatypes_variables,
 )
+from galaxy.work.context import SessionRequestContext
 
 log = logging.getLogger(__name__)
 
@@ -60,6 +68,7 @@ class CreateInstancePayload(BaseModel):
     template_version: int
     variables: SuppliedVariables
     secrets: SuppliedSecrets
+    uuid: Optional[str] = None
 
 
 class UpdateInstancePayload(BaseModel):
@@ -109,16 +118,28 @@ def __init__(self, instance: HasConfigTemplate, payload: TestUpdateInstancePaylo
         self.payload = payload
 
 
+class CreateTestTarget:
+    payload: CreateInstancePayload
+    instance_class: Type[HasConfigSecrets]
+
+    def __init__(self, payload: CreateInstancePayload, instance_class: Type[HasConfigSecrets]):
+        self.payload = payload
+        self.instance_class = instance_class
+
+
 ModifyInstancePayload = Union[UpdateInstanceSecretPayload, UpgradeInstancePayload, UpdateInstancePayload]
 TestModifyInstancePayload = Union[TestUpgradeInstancePayload, TestUpdateInstancePayload]
-CanTestPluginStatus = Union[HasConfigTemplate, CreateInstancePayload, UpgradeTestTarget, UpdateTestTarget]
+CanTestPluginStatus = Union[HasConfigTemplate, CreateTestTarget, UpgradeTestTarget, UpdateTestTarget]
 
 
 def recover_secrets(
-    user_object_store: HasConfigSecrets, vault: Vault, app_config: UsesTemplatesAppConfig
+    user_object_store: HasConfigSecrets, vault: Union[UserVaultWrapper, Vault], app_config: UsesTemplatesAppConfig
 ) -> SecretsDict:
-    user: User = user_object_store.user
-    user_vault = UserVaultWrapper(vault, user)
+    if isinstance(vault, UserVaultWrapper):
+        user_vault = vault
+    else:
+        user: User = user_object_store.user
+        user_vault = UserVaultWrapper(vault, user)
     secrets: SecretsDict = {}
     # now we could recover the list of secrets to fetch from...
     # ones recorded as written in the persisted object, the ones
@@ -138,27 +159,67 @@ class TemplateParameters(TypedDict):
     variables: SuppliedVariables
     environment: EnvironmentDict
     user_details: Dict[str, Any]
+    implicit: Optional[ImplicitConfigurationParameters]
+
+
+class TemplateServerConfiguration:
+    """Extra configuration relevant to a given template.
+
+    For state defined in the server but not defined in the template JSON/YAML directly.
+    Currently this only contains optional oauth2 client credentials and oauth2 configuration
+    information (provider URLs and provider specific configuration for the oauth2 flow).
+    """
+
+    oauth2_client_pair: Optional[OAuth2ClientPair]
+    oauth2_configuration: Optional[OAuth2Configuration]
+
+    def __init__(
+        self,
+        oauth2_client_pair: Optional[OAuth2ClientPair] = None,
+        oauth2_configuration: Optional[OAuth2Configuration] = None,
+        oauth2_scope: Optional[List[str]] = None,
+    ):
+        self.oauth2_client_pair = oauth2_client_pair
+        self.oauth2_configuration = oauth2_configuration
+        self.oauth2_scope = oauth2_scope
+
+    @property
+    def uses_oauth2(self):
+        return self.oauth2_configuration is not None
 
 
 def prepare_template_parameters_for_testing(
     trans: ProvidesUserContext,
     template: Template,
+    template_server_configuration: TemplateServerConfiguration,
     target: CanTestPluginStatus,
     vault: Vault,
     app_config: UsesTemplatesAppConfig,
 ) -> TemplateParameters:
-    secrets = _secrets_for_plugin_status_test(target, vault, app_config)
+    secrets = _secrets_for_plugin_status_test(target, trans.user_vault, app_config)
     variables = _variables_for_plugin_status_test(target)
     environment = prepare_environment_from_root(template.environment, vault, app_config)
     user_details = trans.user.config_template_details()
+    implicit = implicit_parameters_for_testing(
+        trans,
+        template_server_configuration,
+        target,
+        app_config,
+    )
     return TemplateParameters(
-        {"secrets": secrets, "variables": variables, "environment": environment, "user_details": user_details}
+        {
+            "secrets": secrets,
+            "variables": variables,
+            "environment": environment,
+            "user_details": user_details,
+            "implicit": implicit,
+        }
     )
 
 
 def _variables_for_plugin_status_test(target: CanTestPluginStatus) -> SuppliedVariables:
-    if isinstance(target, CreateInstancePayload):
-        return target.variables
+    if isinstance(target, CreateTestTarget):
+        return target.payload.variables
     elif isinstance(target, UpgradeTestTarget) or isinstance(target, UpdateTestTarget):
         if target.instance.template_variables:
             variables = target.instance.template_variables.copy()
@@ -173,21 +234,23 @@ def _variables_for_plugin_status_test(target: CanTestPluginStatus) -> SuppliedVa
 
 
 def _secrets_for_plugin_status_test(
-    target: CanTestPluginStatus, vault: Vault, app_config: UsesTemplatesAppConfig
+    target: CanTestPluginStatus, user_vault: UserVaultWrapper, app_config: UsesTemplatesAppConfig
 ) -> SecretsDict:
-    if isinstance(target, CreateInstancePayload):
-        return target.secrets
+    if isinstance(target, CreateTestTarget):
+        payload = target.payload
+        secrets = payload.secrets.copy()
+        return secrets
     elif isinstance(target, UpgradeTestTarget):
-        secrets = recover_secrets(target.instance, vault, app_config)
+        secrets = recover_secrets(target.instance, user_vault, app_config)
         new_secrets = target.payload.secrets or {}
         for new_secret in new_secrets:
             secrets[new_secret] = new_secrets[new_secret]
         return secrets
     elif isinstance(target, UpdateTestTarget):
-        secrets = recover_secrets(target.instance, vault, app_config)
+        secrets = recover_secrets(target.instance, user_vault, app_config)
         return secrets
     else:
-        secrets = recover_secrets(target, vault, app_config)
+        secrets = recover_secrets(target, user_vault, app_config)
         return secrets
 
 
@@ -375,3 +438,93 @@ def sort_templates(config, catalog: List[T], instance: HasConfigTemplate) -> Lis
     else:
         templates = [stored_template]
     return templates
+
+
+def implicit_parameters_for_testing(
+    trans: ProvidesUserContext,
+    template_server_configuration: TemplateServerConfiguration,
+    target: CanTestPluginStatus,
+    app_config: UsesTemplatesAppConfig,
+) -> Optional[ImplicitConfigurationParameters]:
+    implicit: ImplicitConfigurationParameters = {}
+    if template_server_configuration.oauth2_configuration:
+        refresh_token_key = None
+        oauth2_refresh_token = None
+        if isinstance(target, CreateTestTarget):
+            if target.payload.uuid:
+                refresh_token_key = target.instance_class.vault_key_from_uuid(
+                    target.payload.uuid, "_oauth2_refresh_token", app_config
+                )
+        else:
+            if isinstance(target, (UpgradeTestTarget, UpdateTestTarget)):
+                instance = target.instance
+            else:
+                instance = target
+            refresh_token_key = instance.vault_key_from_uuid(instance.uuid, "_oauth2_refresh_token", app_config)
+        oauth2_refresh_token = trans.user_vault.read_secret(refresh_token_key)
+        if not oauth2_refresh_token:
+            raise InconsistentDatabase(
+                f"Failed to recover oauth2 refresh token from vault at location {refresh_token_key}, Galaxy is in an inconsistent state and probably requires admin intervention"
+            )
+        _inject_oauth2_access_token(implicit, oauth2_refresh_token, template_server_configuration)
+
+    return implicit
+
+
+def implicit_parameters_for_instance(
+    user_instance: HasConfigSecrets,
+    template_server_configuration: TemplateServerConfiguration,
+    vault: Vault,
+    app_config: UsesTemplatesAppConfig,
+) -> ImplicitConfigurationParameters:
+    implicit: ImplicitConfigurationParameters = {}
+
+    if template_server_configuration.oauth2_configuration:
+        refresh_token_key = user_instance.__class__.vault_key_from_uuid(
+            str(user_instance.uuid), "_oauth2_refresh_token", app_config
+        )
+        user_vault = UserVaultWrapper(vault, user_instance.user)
+        oauth2_refresh_token = user_vault.read_secret(refresh_token_key)
+        _inject_oauth2_access_token(implicit, oauth2_refresh_token, template_server_configuration)
+
+    return implicit
+
+
+def _inject_oauth2_access_token(
+    implicit: ImplicitConfigurationParameters,
+    oauth2_refresh_token: str,
+    template_server_configuration: TemplateServerConfiguration,
+) -> None:
+    oauth2_client_pair = template_server_configuration.oauth2_client_pair
+    oauth2_configuration = template_server_configuration.oauth2_configuration
+    assert oauth2_client_pair
+    assert oauth2_configuration
+    response = config_templates.get_token_from_refresh_raw(
+        oauth2_refresh_token, oauth2_client_pair, oauth2_configuration
+    )
+    response.raise_for_status()
+    implicit["oauth2_access_token"] = response.json()["access_token"]
+
+
+def oauth2_refresh_token_status(
+    template_server_configuration: TemplateServerConfiguration, exception: Optional[Exception]
+) -> Optional[PluginAspectStatus]:
+    if not template_server_configuration.uses_oauth2:
+        # no oauth enabled, don't report a status associated with
+        return None
+    else:
+        if not exception:
+            return PluginAspectStatus(
+                state="ok", message="OAuth2 secret refresh token generated an access token for this resource"
+            )
+        else:
+            return PluginAspectStatus(
+                state="not_ok",
+                message=f"OAuth2 secret refresh token failed to generate an access token for this resource: {exception}",
+            )
+
+
+def oauth2_redirect_uri(trans: SessionRequestContext) -> str:
+    galaxy_root = trans.request.url_path
+    redirect_uri = f"{galaxy_root}oauth2_callback"
+    return redirect_uri
diff --git a/lib/galaxy/managers/file_source_instances.py b/lib/galaxy/managers/file_source_instances.py
index 80908ca8728c..81434c9793d5 100644
--- a/lib/galaxy/managers/file_source_instances.py
+++ b/lib/galaxy/managers/file_source_instances.py
@@ -18,8 +18,11 @@
 )
 
 from galaxy.exceptions import (
+    Conflict,
     ItemOwnershipException,
+    ObjectNotFound,
     RequestParameterInvalidException,
+    RequestParameterMissingException,
 )
 from galaxy.files import (
     FileSourceScore,
@@ -44,6 +47,7 @@
     FileSourceTemplate,
     FileSourceTemplateSummaries,
     FileSourceTemplateType,
+    get_oauth2_config_or_none,
     template_to_configuration,
 )
 from galaxy.managers.context import ProvidesUserContext
@@ -52,11 +56,16 @@
     UserFileSource,
 )
 from galaxy.model.scoped_session import galaxy_scoped_session
+from galaxy.schema.schema import OAuth2State
 from galaxy.security.vault import Vault
+from galaxy.util import config_templates  # defer acess to a couple methods to allow monkeypatching in testing
 from galaxy.util.config_templates import (
     connection_exception_to_status,
+    get_authorize_url,
+    OAuth2Info,
     PluginAspectStatus,
     PluginStatus,
+    read_oauth2_info_from_configuration,
     settings_exception_to_status,
     status_template_definition,
     TemplateReference,
@@ -65,16 +74,24 @@
     validate_no_extra_variables_defined,
 )
 from galaxy.util.plugin_config import plugin_source_from_dict
+from galaxy.work.context import SessionRequestContext
 from ._config_templates import (
     CanTestPluginStatus,
     CreateInstancePayload,
+    CreateTestTarget,
+    implicit_parameters_for_instance,
     ModifyInstancePayload,
+    oauth2_redirect_uri,
+    oauth2_refresh_token_status,
     prepare_environment,
+    prepare_environment_from_root,
     prepare_template_parameters_for_testing,
     purge_template_instance,
     recover_secrets,
     save_template_instance,
     sort_templates,
+    TemplateParameters,
+    TemplateServerConfiguration,
     TestModifyInstancePayload,
     TestUpdateInstancePayload,
     TestUpgradeInstancePayload,
@@ -147,6 +164,61 @@ def __init__(
     def summaries(self) -> FileSourceTemplateSummaries:
         return self._catalog.summaries
 
+    def template_oauth2(self, trans: SessionRequestContext, template_id: str, template_version: int) -> OAuth2Info:
+        redirect_uri = oauth2_redirect_uri(trans)
+        template_server_configuration = self._resolver.template_server_configuration(
+            trans.user, template_id, template_version
+        )
+        if template_server_configuration.oauth2_client_pair is None:
+            raise ObjectNotFound(f"Could not find oauth2 configuration for specified plugin template id {template_id}")
+
+        state = OAuth2State(
+            route=f"file_source_instances/{template_id}/{template_version}",
+            nonce="TODO",
+        ).encode()
+        return OAuth2Info(
+            authorize_url=get_authorize_url(
+                template_server_configuration.oauth2_client_pair,
+                template_server_configuration.oauth2_configuration,
+                redirect_uri=redirect_uri,
+                state=state,
+                scope=template_server_configuration.oauth2_scope,
+            )
+        )
+
+    def handle_authorization_code(self, trans: SessionRequestContext, code: str, state: OAuth2State) -> str:
+        route = state.route
+        _, template_id, template_version = route.split("/", 3)
+        redirect_uri = oauth2_redirect_uri(trans)
+        template_server_configuration = self._resolver.template_server_configuration(
+            trans.user, template_id, int(template_version)
+        )
+        if template_server_configuration.oauth2_client_pair is None:
+            raise ObjectNotFound(f"Could not find oauth2 configuration for specified plugin template id {template_id}")
+
+        response = config_templates.get_token_from_code_raw(
+            code,
+            template_server_configuration.oauth2_client_pair,
+            template_server_configuration.oauth2_configuration,
+            redirect_uri,
+        )
+        response.raise_for_status()  # DO BETTER!
+        user_vault = trans.user_vault
+        new_object_uuid = str(uuid4().hex)
+        refresh_token = response.json()["refresh_token"]
+        refresh_token_vault_key = UserFileSource.vault_key_from_uuid(
+            new_object_uuid, "_oauth2_refresh_token", self._app_config
+        )
+        if not refresh_token:
+            raise RequestParameterMissingException("Missing refresh_token in response from remote resource")
+        user_vault.write_secret(refresh_token_vault_key, refresh_token)
+        return f"file_source_templates/{template_id}/new?uuid={new_object_uuid}"
+
+    def _redirect_uri(self, trans: SessionRequestContext) -> str:
+        galaxy_root = trans.request.url_path
+        redirect_uri = f"{galaxy_root}/oauth2_callback"
+        return redirect_uri
+
     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]
@@ -220,7 +292,12 @@ def create_instance(self, trans: ProvidesUserContext, payload: CreateInstancePay
         persisted_file_source = UserFileSource()
         persisted_file_source.user_id = trans.user.id
         assert persisted_file_source.user_id
-        persisted_file_source.uuid = uuid4().hex
+        if payload.uuid:
+            # created a UUID during oauth2 callback to save refresh token
+            self._ensure_no_object_with_uuid(payload.uuid)
+            persisted_file_source.uuid = payload.uuid
+        else:
+            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
@@ -273,13 +350,14 @@ def plugin_status_for_instance(self, trans: ProvidesUserContext, id: str):
         return self._plugin_status(trans, persisted_file_source, to_template_reference(persisted_file_source))
 
     def plugin_status(self, trans: ProvidesUserContext, payload: CreateInstancePayload) -> PluginStatus:
-        return self._plugin_status(trans, payload, payload)
+        target = CreateTestTarget(payload, UserFileSource)
+        return self._plugin_status(trans, target, payload)
 
     def _plugin_status(
-        self, trans: ProvidesUserContext, payload: CanTestPluginStatus, template_reference: TemplateReference
+        self, trans: ProvidesUserContext, target: CanTestPluginStatus, template_reference: TemplateReference
     ):
         template = self._catalog.find_template(template_reference)
-        return self._plugin_status_for_template(trans, payload, template)
+        return self._plugin_status_for_template(trans, target, template)
 
     def _plugin_status_for_template(
         self, trans: ProvidesUserContext, payload: CanTestPluginStatus, template: FileSourceTemplate
@@ -289,7 +367,17 @@ def _plugin_status_for_template(
         if template_definition_status.is_not_ok:
             return PluginStatus(**status_kwds)
         assert template
-        configuration, template_settings_status = self._template_settings_status(trans, payload, template)
+        template_parameters, oauth2_access_token_generation_status = self._template_parameter_generation(
+            trans, payload, template
+        )
+        status_kwds["oauth2_access_token_generation"] = oauth2_access_token_generation_status
+        if not template_parameters or (
+            oauth2_access_token_generation_status and oauth2_access_token_generation_status.is_not_ok
+        ):
+            return PluginStatus(**status_kwds)
+        configuration, template_settings_status = self._template_settings_status(
+            trans, payload, template, template_parameters
+        )
         status_kwds["template_settings"] = template_settings_status
         if template_settings_status.is_not_ok:
             return PluginStatus(**status_kwds)
@@ -304,15 +392,35 @@ def _plugin_status_for_template(
         # locally.
         return PluginStatus(**status_kwds)
 
+    def _template_parameter_generation(
+        self,
+        trans: ProvidesUserContext,
+        payload: CanTestPluginStatus,
+        template: FileSourceTemplate,
+    ) -> Tuple[Optional[TemplateParameters], Optional[PluginAspectStatus]]:
+        template_server_configuration = self._resolver.template_server_configuration(
+            trans.user, template.id, template.version
+        )
+        template_parameters = None
+        exception = None
+        try:
+            template_parameters = prepare_template_parameters_for_testing(
+                trans, template, template_server_configuration, payload, self._app_vault, self._app_config
+            )
+        except Exception as e:
+            if template_server_configuration.uses_oauth2:
+                exception = e
+            else:
+                raise
+        return template_parameters, oauth2_refresh_token_status(template_server_configuration, exception)
+
     def _template_settings_status(
         self,
         trans: ProvidesUserContext,
         payload: CanTestPluginStatus,
         template: FileSourceTemplate,
+        template_parameters: TemplateParameters,
     ) -> Tuple[Optional[FileSourceConfiguration], PluginAspectStatus]:
-        template_parameters = prepare_template_parameters_for_testing(
-            trans, template, payload, self._app_vault, self._app_config
-        )
         configuration = None
         exception = None
         try:
@@ -322,16 +430,19 @@ def _template_settings_status(
         return configuration, settings_exception_to_status(exception)
 
     def _connection_status(
-        self, trans: ProvidesUserContext, payload: CanTestPluginStatus, configuration: FileSourceConfiguration
+        self, trans: ProvidesUserContext, target: CanTestPluginStatus, configuration: FileSourceConfiguration
     ) -> Tuple[Optional[BaseFilesSource], PluginAspectStatus]:
         file_source = None
         exception = None
-        if isinstance(payload, (UpgradeTestTarget, UpdateTestTarget)):
-            label = payload.instance.name
-            doc = payload.instance.description
+        if isinstance(target, (UpgradeTestTarget, UpdateTestTarget)):
+            label = target.instance.name
+            doc = target.instance.description
+        elif isinstance(target, (CreateTestTarget)):
+            label = target.payload.name
+            doc = target.payload.description
         else:
-            label = payload.name
-            doc = payload.description
+            label = target.name
+            doc = target.description
         try:
             file_source_properties = configuration_to_file_source_properties(
                 configuration,
@@ -354,6 +465,12 @@ def _connection_status(
     def _index_filter(self, uuid: str):
         return UserFileSource.__table__.c.uuid == uuid
 
+    def _ensure_no_object_with_uuid(self, uuid: str):
+        filter = UserFileSource.__table__.c.uuid == uuid
+        user_file_source = self._sa_session.query(UserFileSource).filter(filter).one_or_none()
+        if user_file_source is not None:
+            raise Conflict()
+
     def _get(self, trans: ProvidesUserContext, uuid: str) -> UserFileSource:
         filter = self._index_filter(uuid)
         user_file_source = self._sa_session.query(UserFileSource).filter(filter).one_or_none()
@@ -443,13 +560,19 @@ def _file_source_properties_from_uri(self, uri: str) -> Optional[FilesSourceProp
     def _file_source_properties(self, user_file_source: UserFileSource) -> FilesSourceProperties:
         secrets = recover_secrets(user_file_source, self._app_vault, self._app_config)
         environment = prepare_environment(user_file_source, self._app_vault, self._app_config)
+        template_server_configuration = self.template_server_configuration(
+            user_file_source.user, user_file_source.template_id, user_file_source.template_version
+        )
+        implicit = implicit_parameters_for_instance(
+            user_file_source, template_server_configuration, self._app_vault, self._app_config
+        )
         templates = sort_templates(
             self._app_config,
             self._catalog.catalog.root,
             user_file_source,
         )
         file_source_configuration: FileSourceConfiguration = user_file_source.file_source_configuration(
-            secrets=secrets, environment=environment, templates=templates
+            secrets=secrets, environment=environment, implicit=implicit, templates=templates
         )
         return configuration_to_file_source_properties(
             file_source_configuration,
@@ -497,6 +620,31 @@ def _all_user_file_source_properties(self, user_context: FileSourcesUserContext)
             all_file_source_properties.append(files_source_properties)
         return all_file_source_properties
 
+    def template_server_configuration(
+        self, user, template_id: str, template_version: int
+    ) -> TemplateServerConfiguration:
+        catalog = self._catalog
+        template = catalog.find_template_by(template_id, template_version)
+        oauth2_configuration = get_oauth2_config_or_none(template)
+        oauth2_scope = None
+        if oauth2_configuration is not None:
+            environment = prepare_environment_from_root(template.environment, self._app_vault, self._app_config)
+            user_details = user.config_template_details()
+            oauth2_client_pair_obj, oauth2_scope = read_oauth2_info_from_configuration(
+                template.configuration, user_details, environment
+            )
+            # most configuration definitions won't include this, the default for the
+            # plugin type is the fallback and the typical case here
+            if oauth2_scope is None:
+                oauth2_scope = oauth2_configuration.scope
+        else:
+            oauth2_client_pair_obj = None
+        return TemplateServerConfiguration(
+            oauth2_client_pair=oauth2_client_pair_obj,
+            oauth2_configuration=oauth2_configuration,
+            oauth2_scope=oauth2_scope,
+        )
+
     def user_file_sources_to_dicts(
         self,
         for_serialization: bool,
diff --git a/lib/galaxy/managers/object_store_instances.py b/lib/galaxy/managers/object_store_instances.py
index 327a4529a213..40abe01e4100 100644
--- a/lib/galaxy/managers/object_store_instances.py
+++ b/lib/galaxy/managers/object_store_instances.py
@@ -56,6 +56,7 @@
 from ._config_templates import (
     CanTestPluginStatus,
     CreateInstancePayload,
+    CreateTestTarget,
     ModifyInstancePayload,
     prepare_environment,
     prepare_template_parameters_for_testing,
@@ -63,6 +64,7 @@
     recover_secrets,
     save_template_instance,
     sort_templates,
+    TemplateServerConfiguration,
     TestModifyInstancePayload,
     TestUpdateInstancePayload,
     TestUpgradeInstancePayload,
@@ -262,13 +264,14 @@ def plugin_status_for_instance(self, trans: ProvidesUserContext, id: str):
         return self._plugin_status(trans, persisted_object_store, to_template_reference(persisted_object_store))
 
     def plugin_status(self, trans: ProvidesUserContext, payload: CreateInstancePayload) -> PluginStatus:
-        return self._plugin_status(trans, payload, payload)
+        target = CreateTestTarget(payload, UserObjectStore)
+        return self._plugin_status(trans, target, payload)
 
     def _plugin_status(
-        self, trans: ProvidesUserContext, payload: CanTestPluginStatus, template_reference: TemplateReference
+        self, trans: ProvidesUserContext, target: CanTestPluginStatus, template_reference: TemplateReference
     ):
         template = self._catalog.find_template(template_reference)
-        return self._plugin_status_for_template(trans, payload, template)
+        return self._plugin_status_for_template(trans, target, template)
 
     def _plugin_status_for_template(
         self, trans: ProvidesUserContext, payload: CanTestPluginStatus, template: ObjectStoreTemplate
@@ -300,7 +303,7 @@ def _template_settings_status(
         template: ObjectStoreTemplate,
     ) -> Tuple[Optional[ObjectStoreConfiguration], PluginAspectStatus]:
         template_parameters = prepare_template_parameters_for_testing(
-            trans, template, payload, self._app_vault, self._app_config
+            trans, template, TemplateServerConfiguration(), payload, self._app_vault, self._app_config
         )
 
         configuration = None
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 586543f0f9c4..b7a87a74055a 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -192,6 +192,7 @@
 )
 from galaxy.util.config_templates import (
     EnvironmentDict,
+    ImplicitConfigurationParameters,
     SecretsDict,
     Template as ConfigTemplate,
     TemplateEnvironment,
@@ -10956,6 +10957,10 @@ class HasConfigSecrets(RepresentById):
     uuid: Mapped[Union[UUID, str]]
     user: Mapped["User"]
 
+    @classmethod
+    def vault_key_from_uuid(clazz, uuid: str, secret: str, app_config: UsesTemplatesAppConfig) -> str:
+        return f"{clazz.secret_config_type}/{str(get_uuid(uuid))}/{secret}"
+
     def vault_id_prefix(self, app_config: UsesTemplatesAppConfig) -> str:
         id_str = str(self.uuid)
         user_vault_id_prefix = f"{self.secret_config_type}/{id_str}"
@@ -11110,7 +11115,11 @@ def template(self) -> FileSourceTemplate:
         return FileSourceTemplate(**self.template_definition or {})
 
     def file_source_configuration(
-        self, secrets: SecretsDict, environment: EnvironmentDict, templates: Optional[List[FileSourceTemplate]] = None
+        self,
+        secrets: SecretsDict,
+        environment: EnvironmentDict,
+        implicit: ImplicitConfigurationParameters,
+        templates: Optional[List[FileSourceTemplate]] = None,
     ) -> FileSourceConfiguration:
         if templates is None:
             templates = [self.template]
@@ -11125,6 +11134,7 @@ def file_source_configuration(
                     secrets=secrets,
                     user_details=user_details,
                     environment=environment,
+                    implicit=implicit,
                 )
             except Exception as e:
                 if first_exception is None:
diff --git a/lib/galaxy/objectstore/templates/models.py b/lib/galaxy/objectstore/templates/models.py
index dcbabb760434..9ea96058a48a 100644
--- a/lib/galaxy/objectstore/templates/models.py
+++ b/lib/galaxy/objectstore/templates/models.py
@@ -17,7 +17,9 @@
 from galaxy.util.config_templates import (
     EnvironmentDict,
     expand_raw_config,
+    ImplicitConfigurationParameters,
     MarkdownContent,
+    merge_implicit_parameters,
     populate_default_variables,
     SecretsDict,
     StrictModel,
@@ -332,10 +334,12 @@ def template_to_configuration(
     secrets: SecretsDict,
     user_details: UserDetailsDict,
     environment: EnvironmentDict,
+    implicit: Optional[ImplicitConfigurationParameters] = None,
 ) -> ObjectStoreConfiguration:
     configuration_template = template.configuration
     populate_default_variables(template.variables, variables)
     raw_config = expand_raw_config(configuration_template, variables, secrets, user_details, environment)
+    merge_implicit_parameters(raw_config, implicit)
     return to_configuration_object(raw_config)
 
 
diff --git a/lib/galaxy/schema/schema.py b/lib/galaxy/schema/schema.py
index e6e92e79334e..80200bbed0af 100644
--- a/lib/galaxy/schema/schema.py
+++ b/lib/galaxy/schema/schema.py
@@ -1,5 +1,6 @@
 """This module contains general pydantic models and common schema field annotations for them."""
 
+import base64
 from datetime import (
     date,
     datetime,
@@ -3720,6 +3721,18 @@ class PageSummary(PageSummaryBase, WithModelClass):
 )
 
 
+class OAuth2State(BaseModel):
+    route: str
+    nonce: str
+
+    def encode(self) -> str:
+        return base64.b64encode(self.model_dump_json().encode("utf-8")).decode("utf-8")
+
+    @staticmethod
+    def decode(base64_param: str) -> "OAuth2State":
+        return OAuth2State.model_validate_json(base64.b64decode(base64_param.encode("utf-8")))
+
+
 class PageDetails(PageSummary):
     content_format: PageContentFormat = ContentFormatField
     content: Optional[str] = ContentField
diff --git a/lib/galaxy/util/config_templates.py b/lib/galaxy/util/config_templates.py
index 79f0bed52ca9..b2319cdcfdd6 100644
--- a/lib/galaxy/util/config_templates.py
+++ b/lib/galaxy/util/config_templates.py
@@ -3,17 +3,22 @@
 This is capturing code shared by file source templates and object store templates.
 """
 
+import logging
 import os
 from typing import (
     Any,
+    cast,
     Dict,
     List,
     Optional,
+    Tuple,
     Type,
     TypeVar,
     Union,
 )
+from urllib.parse import urlencode
 
+import requests
 import yaml
 from boltons.iterutils import remap
 from pydantic import (
@@ -24,7 +29,9 @@
 )
 from typing_extensions import (
     Literal,
+    NotRequired,
     Protocol,
+    TypedDict,
 )
 
 try:
@@ -45,6 +52,8 @@
 )
 from galaxy.util import asbool
 
+log = logging.getLogger(__name__)
+
 TemplateVariableType = Literal["string", "path_component", "boolean", "integer"]
 TemplateVariableValueType = Union[str, bool, int]
 TemplateExpansion = str
@@ -178,6 +187,12 @@ def expand_raw_config(
         "environment": environment,
     }
 
+    return _expand_raw_config(template_configuration, template_variables)
+
+
+def _expand_raw_config(
+    template_configuration: TemplateConfiguration, template_variables: Dict[str, Any]
+) -> RawTemplateConfig:
     template_start = template_configuration.template_start or "{{"
     template_end = template_configuration.template_end or "}}"
 
@@ -193,6 +208,14 @@ def expand_template(_, key, value):
     return raw_config
 
 
+def merge_implicit_parameters(raw_config: RawTemplateConfig, implicit: Optional["ImplicitConfigurationParameters"]):
+    if implicit:
+        raw_config.update(implicit)
+        raw_config.pop("oauth2_client_id", None)
+        raw_config.pop("oauth2_client_secret", None)
+        raw_config.pop("oauth2_scope", None)
+
+
 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)
@@ -283,6 +306,9 @@ def id(self) -> str: ...
     @property
     def version(self) -> int: ...
 
+    @property
+    def type(self) -> str: ...
+
     @property
     def variables(self) -> Optional[List[TemplateVariable]]: ...
 
@@ -445,6 +471,7 @@ class PluginStatus(StrictModel):
 
     # TODO: Fill in writable checks.
     # writable: Optional[PluginAspectStatus] = None
+    oauth2_access_token_generation: Optional[PluginAspectStatus] = None
 
 
 def status_template_definition(template: Optional[Template]) -> PluginAspectStatus:
@@ -478,3 +505,106 @@ def connection_exception_to_status(what: str, exception: Optional[Exception]) ->
         message = f"Failed to connect to a {what} with supplied settings: {exception}"
         connection_status = PluginAspectStatus(state="not_ok", message=message)
     return connection_status
+
+
+class OAuth2Info(StrictModel):
+    authorize_url: str
+
+
+class OAuth2Configuration(StrictModel):
+    authorize_url: str
+    token_url: str
+    authorize_params: Optional[Dict[str, str]]
+    scope: Optional[List[str]] = None
+
+
+ConfiguredOAuth2Sources = Dict[str, OAuth2Configuration]
+
+
+class OAuth2ClientPair(StrictModel):
+    client_id: str
+    client_secret: str
+
+
+def get_authorize_url(
+    client_id_or_pair: Union[str, OAuth2ClientPair],
+    config: OAuth2Configuration,
+    redirect_uri: Optional[str],
+    state: Optional[str] = None,
+    scope: Optional[List[str]] = None,
+) -> str:
+    client_id = client_id_or_pair if isinstance(client_id_or_pair, str) else client_id_or_pair.client_id
+    query_data = dict(
+        client_id=client_id,
+        response_type="code",
+    )
+    if redirect_uri is not None:
+        query_data["redirect_uri"] = redirect_uri
+    if state is not None:
+        query_data["state"] = state
+    if scope is not None:
+        query_data["scope"] = ",".join(scope)
+    elif config.scope is not None:
+        query_data["scope"] = ",".join(config.scope)
+    query_data.update(config.authorize_params or {})
+    query = urlencode(query_data)
+    return f"{config.authorize_url}?{query}"
+
+
+def get_token_from_code_raw(
+    code: str, client_pair: OAuth2ClientPair, config: OAuth2Configuration, redirect_uri: Optional[str]
+) -> requests.Response:
+    data = {
+        "code": code,
+        "grant_type": "authorization_code",
+        "client_id": client_pair.client_id,
+        "client_secret": client_pair.client_secret,
+    }
+    if redirect_uri is not None:
+        data["redirect_uri"] = redirect_uri
+
+    return requests.post(config.token_url, data=data)
+
+
+def get_token_from_refresh_raw(
+    refresh_token: str, client_pair: OAuth2ClientPair, config: OAuth2Configuration
+) -> requests.Response:
+    data = {
+        "refresh_token": refresh_token,
+        "grant_type": "refresh_token",
+        "client_id": client_pair.client_id,
+        "client_secret": client_pair.client_secret,
+    }
+
+    return requests.post(config.token_url, data=data)
+
+
+def get_oauth2_config_from(template, sources: ConfiguredOAuth2Sources) -> OAuth2Configuration:
+    template_type = template.configuration.type
+    if template_type not in sources:
+        raise ObjectNotFound(f"oauth information not available for template type {template_type}")
+    return sources[template_type]
+
+
+def read_oauth2_info_from_configuration(
+    template_configuration: TemplateConfiguration,
+    user_details: UserDetailsDict,
+    environment: EnvironmentDict,
+) -> Tuple[OAuth2ClientPair, Optional[str]]:
+    template_variables = {
+        "user": user_details,
+        "environment": environment,
+    }
+
+    expanded_config = _expand_raw_config(template_configuration, template_variables)
+    oauth2_client_id = expanded_config["oauth2_client_id"]
+    oauth2_client_secret = expanded_config["oauth2_client_secret"]
+    oauth2_scope = cast(Optional[str], expanded_config.get("oauth2_scope"))
+    client_pair = OAuth2ClientPair(client_id=oauth2_client_id, client_secret=oauth2_client_secret)
+    return client_pair, oauth2_scope
+
+
+# Things added to configuration dictionary not managed by the template
+# but injected dynamically. Currently only `oauth2_access_token`.
+class ImplicitConfigurationParameters(TypedDict):
+    oauth2_access_token: NotRequired[str]
diff --git a/lib/galaxy/webapps/galaxy/api/file_sources.py b/lib/galaxy/webapps/galaxy/api/file_sources.py
index ee7816714822..4e9a510c9a0b 100644
--- a/lib/galaxy/webapps/galaxy/api/file_sources.py
+++ b/lib/galaxy/webapps/galaxy/api/file_sources.py
@@ -17,7 +17,11 @@
     TestModifyInstancePayload,
     UserFileSourceModel,
 )
-from galaxy.util.config_templates import PluginStatus
+from galaxy.util.config_templates import (
+    OAuth2Info,
+    PluginStatus,
+)
+from galaxy.work.context import SessionRequestContext
 from . import (
     depends,
     DependsOnTrans,
@@ -33,6 +37,14 @@
     ..., title="User File Source UUID", description="The UUID index for a persisted UserFileSourceStore object."
 )
 
+TemplateIdPathParam: str = Path(
+    ..., title="Template ID", description="The template ID of the target file source template."
+)
+
+TemplateVersionPathParam = Path(
+    ..., title="Template Version", description="The template version of the target file source template."
+)
+
 
 @router.cbv
 class FastAPIFileSources:
@@ -50,6 +62,19 @@ def index_templates(
     ) -> FileSourceTemplateSummaries:
         return self.file_source_instances_manager.summaries
 
+    @router.get(
+        "/api/file_source_templates/{template_id}/{template_version}/oauth2",
+        response_description="OAuth2 authorization url to redirect user to prior to creation.",
+        operation_id="file_sources__template_oauth2",
+    )
+    def template_oauth2(
+        self,
+        trans: SessionRequestContext = DependsOnTrans,
+        template_id: str = TemplateIdPathParam,
+        template_version: int = TemplateVersionPathParam,
+    ) -> OAuth2Info:
+        return self.file_source_instances_manager.template_oauth2(trans, template_id, template_version)
+
     @router.post(
         "/api/file_source_instances",
         summary="Create a user-bound file source.",
diff --git a/lib/galaxy/webapps/galaxy/api/oauth2_callback.py b/lib/galaxy/webapps/galaxy/api/oauth2_callback.py
new file mode 100644
index 000000000000..bb7968d89c58
--- /dev/null
+++ b/lib/galaxy/webapps/galaxy/api/oauth2_callback.py
@@ -0,0 +1,50 @@
+from fastapi import Query
+from fastapi.responses import RedirectResponse
+
+from galaxy.exceptions import ObjectNotFound
+from galaxy.managers.file_source_instances import FileSourceInstancesManager
+from galaxy.schema.schema import OAuth2State
+from galaxy.webapps.galaxy.api import (
+    depends,
+    DependsOnTrans,
+    Router,
+)
+from galaxy.work.context import SessionRequestContext
+
+StateQueryParam: str = Query(
+    ...,
+    title="State information sent with auth request",
+    description="Base-64 encoded JSON used to route request within Galaxy.",
+)
+CodeQueryParam: str = Query(
+    ...,
+    title="OAuth2 Authorization Code from remote resource",
+)
+
+router = Router(tags=["oauth2"])
+
+
+@router.cbv
+class OAuth2Callback:
+    file_source_instances_manager: FileSourceInstancesManager = depends(FileSourceInstancesManager)
+
+    @router.get(
+        "/oauth2_callback",
+        summary="Callback entry point for remote resource responses with OAuth2 authorization codes",
+    )
+    def oauth2_callback(
+        self,
+        trans: SessionRequestContext = DependsOnTrans,
+        state: str = StateQueryParam,
+        code: str = CodeQueryParam,
+    ):
+        oauth2_state = OAuth2State.decode(state)
+        # TODO: save session information in cookie to verify not CSRF with oauth2_state.nonce
+        route = oauth2_state.route
+        if route.startswith("file_source_instances/"):
+            redirect = self.file_source_instances_manager.handle_authorization_code(trans, code, oauth2_state)
+        # implement other routes here as needed for other components
+        else:
+            raise ObjectNotFound(f"Could not find oauth2 callback for route {route}")
+
+        return RedirectResponse(f"{trans.request.url_path}{redirect}")
diff --git a/lib/galaxy/work/context.py b/lib/galaxy/work/context.py
index 025fb7aac920..57ae49a99a20 100644
--- a/lib/galaxy/work/context.py
+++ b/lib/galaxy/work/context.py
@@ -94,6 +94,11 @@ class GalaxyAbstractRequest:
     def base(self) -> str:
         """Base URL of the request."""
 
+    @property
+    @abc.abstractmethod
+    def url_path(self) -> str:
+        """Base with optional prefix added."""
+
     @property
     @abc.abstractmethod
     def host(self) -> str:
diff --git a/test/unit/app/managers/test_user_file_sources.py b/test/unit/app/managers/test_user_file_sources.py
index 78a3faef6aaf..12ccb7d16b1c 100644
--- a/test/unit/app/managers/test_user_file_sources.py
+++ b/test/unit/app/managers/test_user_file_sources.py
@@ -5,7 +5,14 @@
     Optional,
     Type,
 )
-
+from unittest import SkipTest
+from uuid import uuid4
+
+from fs.osfs import OSFS
+try:
+    from fs.dropboxfs import DropboxFS
+except ImportError:
+    DropboxFS = None
 from yaml import safe_load
 
 from galaxy.exceptions import (
@@ -13,6 +20,7 @@
     RequestParameterMissingException,
 )
 from galaxy.files import FileSourcesUserContext
+from galaxy.files.sources import dropbox
 from galaxy.files.templates import ConfiguredFileSourceTemplates
 from galaxy.files.templates.examples import get_example
 from galaxy.managers.file_source_instances import (
@@ -29,6 +37,12 @@
     UserDefinedFileSourcesImpl,
     UserFileSourceModel,
 )
+from galaxy.model import (
+    get_uuid,
+    UserFileSource,
+)
+from galaxy.schema.schema import OAuth2State
+from galaxy.util import config_templates
 from galaxy.util.config_templates import RawTemplateConfig
 from .base import BaseTestCase
 
@@ -207,6 +221,138 @@ def test_create_posix(self, tmp_path):
         assert file_source.get_scheme() == USER_FILE_SOURCES_SCHEME
         assert file_source.get_uri_root() == uri_root
 
+    def test_oauth2_flow(self, tmp_path, monkeypatch):
+
+        json = {
+            "refresh_token": "my_test_refresh_token",
+        }
+
+        def mock_get_token_from_code_raw(
+            code,
+            client_pair,
+            config,
+            redirect_uri,
+        ):
+            return MockResponse(json)
+
+        monkeypatch.setattr(config_templates, "get_token_from_code_raw", mock_get_token_from_code_raw)
+
+        self._init_dropbox_env(tmp_path, monkeypatch)
+
+        authorize_url = self.manager.template_oauth2(self.trans, "dropbox", 0).authorize_url
+        from urllib.parse import (
+            parse_qs,
+            urlparse,
+        )
+
+        parse_result = urlparse(authorize_url)
+        assert parse_result.hostname == "www.dropbox.com"
+        assert parse_result.path == "/oauth2/authorize"
+        query_params = parse_qs(parse_result.query)
+        assert "state" in query_params
+        state_param = query_params["state"]
+        state = OAuth2State.decode(state_param[0])
+        assert state.route == "file_source_instances/dropbox/0"
+        redirect_url = self.manager.handle_authorization_code(
+            self.trans,
+            "moocow",
+            state,
+        )
+        parse_result = urlparse(redirect_url)
+        query_params = parse_qs(parse_result.query)
+        assert "uuid" in query_params
+        uuid_param = query_params["uuid"]
+        uuid = uuid_param[0]
+
+        user_vault = self.trans.user_vault
+        config_secret_key = UserFileSource.vault_key_from_uuid(uuid, "_oauth2_refresh_token", None)
+        assert user_vault.read_secret(config_secret_key)
+
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="dropbox",
+            template_version=0,
+            variables={},
+            secrets={},
+            uuid=uuid,
+        )
+        user_object = self._create_instance(create_payload)
+        assert get_uuid(user_object.uuid) == get_uuid(uuid)
+
+    def test_oauth2_access_token_injection_during_verify(self, tmp_path, monkeypatch):
+        if DropboxFS is None:
+            raise SkipTest("Optional dropbpox dependency not available")
+        self._init_dropbox_env(tmp_path, monkeypatch)
+
+        uuid = uuid4().hex
+        user_vault = self.trans.user_vault
+        config_secret_key = UserFileSource.vault_key_from_uuid(uuid, "_oauth2_refresh_token", None)
+        user_vault.write_secret(config_secret_key, "test_refresh_token")
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="dropbox",
+            template_version=0,
+            variables={},
+            secrets={},
+            uuid=uuid,
+        )
+        self._create_instance(create_payload)
+        json = {
+            "access_token": "my_test_access_token",
+        }
+
+        def mock_get_token_from_refresh_raw(refresh_token, client_pair, config):
+            return MockResponse(json)
+
+        pyfilesystem_fs_init_kwd = {}
+
+        class MockDropboxFS(OSFS):
+
+            def __init__(self, **kwd):
+                pyfilesystem_fs_init_kwd.update(kwd)
+                root = tmp_path / "foobar"
+                root.mkdir()
+                super().__init__(root)
+
+        monkeypatch.setattr(config_templates, "get_token_from_refresh_raw", mock_get_token_from_refresh_raw)
+        monkeypatch.setattr(dropbox, "DropboxFS", MockDropboxFS)
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert status.oauth2_access_token_generation
+        assert not status.oauth2_access_token_generation.is_not_ok
+        assert status.connection
+        assert not status.connection.is_not_ok
+        assert pyfilesystem_fs_init_kwd["access_token"] == "my_test_access_token"
+
+    def test_report_oauth2_access_token_generation_failure(self, tmp_path, monkeypatch):
+        self._init_dropbox_env(tmp_path, monkeypatch)
+
+        uuid = uuid4().hex
+        user_vault = self.trans.user_vault
+        config_secret_key = UserFileSource.vault_key_from_uuid(uuid, "_oauth2_refresh_token", None)
+        user_vault.write_secret(config_secret_key, "test_refresh_token")
+        create_payload = CreateInstancePayload(
+            name=SIMPLE_FILE_SOURCE_NAME,
+            description=SIMPLE_FILE_SOURCE_DESCRIPTION,
+            template_id="dropbox",
+            template_version=0,
+            variables={},
+            secrets={},
+            uuid=uuid,
+        )
+        self._create_instance(create_payload)
+        excepton_message = "There is an issue with the refresh_token"
+
+        def mock_get_token_from_refresh_raw(refresh_token, client_pair, config):
+            return MockExceptionResponse(excepton_message)
+
+        monkeypatch.setattr(config_templates, "get_token_from_refresh_raw", mock_get_token_from_refresh_raw)
+        status = self.manager.plugin_status(self.trans, create_payload)
+        assert status.oauth2_access_token_generation
+        assert status.oauth2_access_token_generation.is_not_ok
+        assert excepton_message in status.oauth2_access_token_generation.message
+
     def test_io(self, tmp_path):
         self._init_managers(tmp_path)
         user_file_source = self._create_user_file_source()
@@ -641,6 +787,13 @@ def _init_and_create_simple(self, tmp_path) -> UserFileSourceModel:
         user_file_source = self._create_user_file_source()
         return user_file_source
 
+    def _init_dropbox_env(self, tmp_path, monkeypatch):
+        self.init_user_in_database()
+        self._init_managers(tmp_path, safe_load(get_example("production_dropbox.yml")))
+
+        monkeypatch.setenv("GALAXY_DROPBOX_APP_CLIENT_ID", "mock_client_id")
+        monkeypatch.setenv("GALAXY_DROPBOX_APP_CLIENT_SECRET", "mock_client_secret")
+
     def _create_user_file_source(self, template_id="home_directory") -> UserFileSourceModel:
         create_payload = CreateInstancePayload(
             name=SIMPLE_FILE_SOURCE_NAME,
@@ -704,3 +857,30 @@ def _init_managers(self, tmp_path, config_dict=None):
         manager = self.app[FileSourceInstancesManager]
         self.file_sources = file_sources
         self.manager = manager
+
+
+class MockResponse:
+    _error_checked = False
+
+    def __init__(self, json):
+        self._json = json
+
+    def raise_for_status(self):
+        self._error_checked = True
+
+    def json(self):
+        return self._json
+
+    def assert_error_checked(self):
+        assert self._error_checked
+
+
+class MockExceptionResponse:
+
+    def __init__(self, exception_msg: str):
+        self._exception_msg = exception_msg
+
+    def raise_for_status(self):
+        from requests.exceptions import HTTPError
+
+        raise HTTPError(self._exception_msg, self._exception_msg)
diff --git a/test/unit/files/test_template_models.py b/test/unit/files/test_template_models.py
index 9d11069fc01d..c23b6bcf943b 100644
--- a/test/unit/files/test_template_models.py
+++ b/test/unit/files/test_template_models.py
@@ -295,11 +295,14 @@ def test_examples_parse():
     _assert_example_parses("production_aws_private_bucket.yml")
     _assert_example_parses("production_aws_public_bucket.yml")
     _assert_example_parses("production_s3fs.yml")
+    _assert_example_parses("production_dropbox.yml")
     _assert_example_parses("s3fs_by_host_and_port.yml")
     _assert_example_parses("templating_override.yml")
     _assert_example_parses("admin_secrets.yml")
     _assert_example_parses("admin_secrets_with_defaults.yml")
     _assert_example_parses("testing_multi_version_with_secrets.yml")
+    _assert_example_parses("dropbox_client_secrets_in_vault.yml")
+    _assert_example_parses("dropbox_client_secrets_explicit.yml")
 
 
 def _assert_example_parses(filename: str):
diff --git a/test/unit/schema/test_schema.py b/test/unit/schema/test_schema.py
index 21b37096d131..07d2cb3b3d51 100644
--- a/test/unit/schema/test_schema.py
+++ b/test/unit/schema/test_schema.py
@@ -5,6 +5,7 @@
 
 from galaxy.schema.schema import (
     DatasetStateField,
+    OAuth2State,
     TAG_ITEM_PATTERN,
 )
 from galaxy.schema.tasks import (
@@ -78,3 +79,10 @@ def test_invalid(self):
         ]
         for t in tag_strings:
             assert not re.match(TAG_ITEM_PATTERN, t)
+
+
+def test_oauth_state():
+    state_in = OAuth2State(route="/file_sources/dropbox", nonce="abcde56")
+    state_out = OAuth2State.decode(state_in.encode())
+    assert state_out.route == "/file_sources/dropbox"
+    assert state_out.nonce == "abcde56"
diff --git a/test/unit/util/test_config_template_validation.py b/test/unit/util/test_config_template_validation.py
index dd388068e9ee..cde93f5063ff 100644
--- a/test/unit/util/test_config_template_validation.py
+++ b/test/unit/util/test_config_template_validation.py
@@ -27,6 +27,7 @@
 
 class TestTemplate(StrictModel):
     id: str
+    type: str = "test"
     version: int
     variables: Optional[List[TemplateVariable]]
     secrets: Optional[List[TemplateSecret]]