diff --git a/.github/workflows/container.yml b/.github/workflows/container.yml
index b736b22e..65f452cd 100644
--- a/.github/workflows/container.yml
+++ b/.github/workflows/container.yml
@@ -11,3 +11,4 @@ jobs:
     with:
       image_namespace: wipacrepo
       image_name: iceprod
+      mode: BUILD
diff --git a/iceprod/credentials/server.py b/iceprod/credentials/server.py
index 8592b5b8..735a89c4 100644
--- a/iceprod/credentials/server.py
+++ b/iceprod/credentials/server.py
@@ -8,6 +8,7 @@
 import time
 from typing import Any
 
+import jwt
 from prometheus_client import Info, start_http_server
 import pymongo
 import pymongo.errors
@@ -24,7 +25,8 @@
 from iceprod.rest.auth import authorization
 from iceprod.rest.base_handler import IceProdRestConfig, APIBase
 from iceprod.server.util import nowstr, datetime2str
-from .service import RefreshService, get_expiration, is_expired
+from .service import RefreshService
+from .util import get_expiration, is_expired
 
 logger = logging.getLogger('server')
 
@@ -74,6 +76,7 @@ async def create(self, db, base_data):
         argo.add_argument('secret_key', type=str, default='', required=False)
         argo.add_argument('access_token', type=str, default='', required=False)
         argo.add_argument('refresh_token', type=str, default='', required=False)
+        argo.add_argument('scope', type=str, default=None, required=False)
         argo.add_argument('expire_date', type=float, default=now, required=False)
         argo.add_argument('last_use', type=float, default=now, required=False)
         args = vars(argo.parse_args())
@@ -103,13 +106,18 @@ async def create(self, db, base_data):
                 raise HTTPError(400, reason='must specify either access or refresh tokens')
             data['access_token'] = args['access_token']
             data['refresh_token'] = args['refresh_token']
+            if args['scope']:
+                base_data['scope'] = args['scope']
+            data['scope'] = args['scope']
             data['expiration'] = args['expire_date']
             data['last_use'] = args['last_use']
+            if data['access_token'] and data['scope'] is None:
+                data['scope'] = jwt.decode(data['access_token'], options={"verify_signature": False}).get('scope', None)
 
-            if 'refresh_token' in data and not data.get('access_token', ''):
+            if 'refresh_token' in data and not data['access_token']:
                 new_cred = await self.refresh_service.refresh_cred(data)
                 data.update(new_cred)
-            if (not data.get('access_token', '')) and data.get('expiration') == now:
+            if (not data['access_token']) and data.get('expiration') == now:
                 data['expiration'] = get_expiration(data['access_token'])
 
         else:
@@ -130,13 +138,14 @@ async def patch_cred(self, db, base_data):
         argo.add_argument('secret_key', type=str, default='', required=False)
         argo.add_argument('access_token', type=str, default='', required=False)
         argo.add_argument('refresh_token', type=str, default='', required=False)
+        argo.add_argument('scope', type=str, default='', required=False)
         argo.add_argument('expiration', type=float, default=0, required=False)
         argo.add_argument('last_use', type=float, default=0, required=False)
         args = vars(argo.parse_args())
         base_data['url'] = args['url']
 
         data = {}
-        for key in ('buckets', 'access_key', 'secret_key', 'access_token', 'refresh_token', 'expiration', 'last_use'):
+        for key in ('buckets', 'access_key', 'secret_key', 'access_token', 'refresh_token', 'scope', 'expiration', 'last_use'):
             if val := args[key]:
                 data[key] = val
 
@@ -157,6 +166,9 @@ async def search_creds(self, db, base_data):
         assert self.refresh_service
         if url := self.get_argument('url', None):
             base_data['url'] = url
+        if scope := self.get_argument('scope', None):
+            base_data['scope'] = scope
+        logger.info('base_data: %r', base_data)
 
         refresh = self.get_argument('norefresh', None) is None
 
@@ -166,23 +178,34 @@ async def search_creds(self, db, base_data):
             filters['type'] = 'oauth'
             await db.update_many(filters, {'$set': update_data})
 
-        ret = {}
+        ret = []
         async for row in db.find(base_data, projection={'_id': False}):
-            ret[row['url']] = row
-
-        for key in list(ret):
-            cred = ret[key]
-            if refresh and is_expired(cred) and cred['refresh_token']:
+            if refresh and is_expired(row) and row['refresh_token']:
                 try:
-                    new_cred = await self.refresh_service.refresh_cred(cred)
+                    new_cred = await self.refresh_service.refresh_cred(row)
                     filters = base_data.copy()
-                    filters['url'] = key
-                    ret[key] = await db.find_one_and_update(filters, {'$set': new_cred}, projection={'_id': False})
+                    filters['url'] = row['url']
+                    cred = await db.find_one_and_update(filters, {'$set': new_cred}, projection={'_id': False})
+                    ret.append(cred)
                 except Exception:
-                    del ret[key]
+                    logging.debug('ignore expired token %r', row)
+            else:
+                ret.append(row)
 
         return ret
 
+    async def delete_cred(self, db, base_data):
+        argo = ArgumentHandler(ArgumentSource.JSON_BODY_ARGUMENTS, self)
+        argo.add_argument('url', type=str, default='', required=False)
+        argo.add_argument('scope', type=str, default=None, required=False)
+        body_args = argo.parse_args()
+        if body_args.url:
+            base_data['url'] = body_args.url
+        if body_args.scope is not None:
+            base_data['scope'] = body_args.scope
+
+        await db.delete_many(base_data)
+
 
 class GroupCredentialsHandler(BaseCredentialsHandler):
     """
@@ -221,6 +244,7 @@ async def post(self, groupname):
         OAuth body args:
             access_token (str): access token
             refresh_token (str): refresh token
+            scope (str): scope of access token
             expire_date (str): access token expiration, ISO date time in UTC (optional)
 
         Args:
@@ -237,8 +261,9 @@ async def patch(self, groupname):
         """
         Update a group credential.  Usually used to update a specifc field.
 
-        Required body args:
+        Body args:
             url (str): url of controlled resource
+            scope (str): (optional) scope of access token
 
         Other body args will update a credential.
 
@@ -260,19 +285,12 @@ async def delete(self, groupname):
             groupname (str): groupname
         Body args:
             url (str): (optional) url of controlled resource
+            scope (str): (optional) scope of access token
         """
         if self.auth_roles == ['user'] and groupname not in self.auth_groups:
             raise HTTPError(403, 'unauthorized')
 
-        args = {'groupname': groupname}
-
-        argo = ArgumentHandler(ArgumentSource.JSON_BODY_ARGUMENTS, self)
-        argo.add_argument('url', type=str, default='', required=False)
-        body_args = argo.parse_args()
-        if body_args.url:
-            args['url'] = body_args.url
-
-        await self.db.group_creds.delete_many(args)
+        await self.delete_cred(self.db.group_creds, {'groupname': groupname})
         self.write({})
 
 
@@ -313,6 +331,7 @@ async def post(self, username):
         OAuth body args:
             access_token (str): access token
             refresh_token (str): refresh token
+            scope (str): scope of access token
             expire_date (str): access token expiration, ISO date time in UTC (optional)
 
         Args:
@@ -329,8 +348,9 @@ async def patch(self, username):
         """
         Update a user credential.  Usually used to update a specifc field.
 
-        Required body args:
+        Body args:
             url (str): url of controlled resource
+            scope (str): (optional) scope of access token
 
         Other body args will update a credential.
 
@@ -352,21 +372,129 @@ async def delete(self, username):
             username (str): username
         Body args:
             url (str): (optional) url of controlled resource
+            scope (str): (optional) scope of access token
         Returns:
             dict: url: credential dict
         """
         if self.auth_roles == ['user'] and username != self.current_user:
             raise HTTPError(403, 'unauthorized')
 
-        args = {'username': username}
+        await self.delete_cred(self.db.user_creds, {'username': username})
+        self.write({})
 
-        argo = ArgumentHandler(ArgumentSource.JSON_BODY_ARGUMENTS, self)
-        argo.add_argument('url', type=str, default='', required=False)
-        body_args = argo.parse_args()
-        if body_args.url:
-            args['url'] = body_args.url
 
-        await self.db.user_creds.delete_many(args)
+class DatasetCredentialsHandler(BaseCredentialsHandler):
+    """
+    Handle dataset credentials requests.
+    """
+    @authorization(roles=['admin', 'system'])
+    async def get(self, dataset_id):
+        """
+        Get a datasets's credentials.
+
+        Args:
+            dataset_id (str): dataset_id
+        Returns:
+            dict: url: credential dict
+        """
+        ret = await self.search_creds(self.db.dataset_creds, {'dataset_id': dataset_id})
+        self.write(ret)
+
+    @authorization(roles=['admin', 'system'])
+    async def delete(self, dataset_id):
+        """
+        Delete a dataset's credentials.
+
+        Args:
+            dataset_id (str): dataset_id
+        Body args:
+            url (str): (optional) url of controlled resource
+            scope (str): (optional) scope of access token
+        Returns:
+            dict: url: credential dict
+        """
+        await self.delete_cred(self.db.dataset_creds, {'dataset_id': dataset_id})
+        self.write({})
+
+
+class DatasetTaskCredentialsHandler(BaseCredentialsHandler):
+    """
+    Handle dataset/task credentials requests.
+    """
+    @authorization(roles=['admin', 'system'])
+    async def get(self, dataset_id, task_name):
+        """
+        Get a datasets's credentials.
+
+        Args:
+            dataset_id (str): dataset_id
+            task_name (str): task name
+        Returns:
+            dict: url: credential dict
+        """
+        ret = await self.search_creds(self.db.dataset_creds, {'dataset_id': dataset_id, 'task_name': task_name})
+        self.write(ret)
+
+    @authorization(roles=['admin', 'system'])
+    async def post(self, dataset_id, task_name):
+        """
+        Set a dataset credential.  Overwrites an existing credential for the specified url.
+
+        Common body args:
+            url (str): url of controlled resource
+            type (str): credential type (`s3` or `oauth`)
+
+        S3 body args:
+            buckets (list): list of buckets for this url, or [] if using virtual-hosted buckets in the url
+            access_key (str): access key
+            secret_key (str): secret key
+
+        OAuth body args:
+            access_token (str): access token
+            refresh_token (str): refresh token
+            scope (str): scope of access token
+            expire_date (str): access token expiration, ISO date time in UTC (optional)
+
+        Args:
+            dataset_id (str): dataset_id
+            task_name (str): task name
+        """
+        await self.create(self.db.dataset_creds, {'dataset_id': dataset_id, 'task_name': task_name})
+        self.write({})
+
+    @authorization(roles=['admin', 'system'])
+    async def patch(self, dataset_id, task_name):
+        """
+        Update a dataset credential.  Usually used to update a specifc field.
+
+        Body args:
+            url (str): url of controlled resource
+            scope (str): (optional) scope of access token
+
+        Other body args will update a credential.
+
+        Args:
+            dataset_id (str): dataset_id
+            task_name (str): task name
+        """
+        await self.patch_cred(self.db.dataset_creds, {'dataset_id': dataset_id, 'task_name': task_name})
+        self.write({})
+
+    @authorization(roles=['admin', 'system'])
+    async def delete(self, dataset_id, task_name):
+        """
+        Delete a dataset's credentials.
+
+        Args:
+            dataset_id (str): dataset_id
+            task_name (str): task name
+        Body args:
+            url (str): (optional) url of controlled resource
+            scope (str): (optional) scope of access token
+        Returns:
+            dict: url: credential dict
+        """
+        await self.delete_cred(self.db.dataset_creds, {'dataset_id': dataset_id, 'task_name': task_name})
         self.write({})
 
 
@@ -483,6 +611,16 @@ def __init__(self):
             },
             'user_creds': {
                 'username_index': {'keys': [('username', pymongo.DESCENDING), ('url', pymongo.DESCENDING)], 'unique': True},
+            },
+            'dataset_creds': {
+                'dataset_index': {
+                    'keys': [
+                        ('dataset_id', pymongo.DESCENDING),
+                        ('task_name', pymongo.DESCENDING),
+                        ('url', pymongo.DESCENDING)
+                    ],
+                    'unique': True,
+                },
             }
         }
 
@@ -516,6 +654,8 @@ def __init__(self):
 
         server.add_route(r'/groups/(?P<groupname>\w+)/credentials', GroupCredentialsHandler, kwargs)
         server.add_route(r'/users/(?P<username>\w+)/credentials', UserCredentialsHandler, kwargs)
+        server.add_route(r'/datasets/(?P<dataset_id>\w+)/credentials', DatasetCredentialsHandler, kwargs)
+        server.add_route(r'/datasets/(?P<dataset_id>\w+)/tasks/(?P<task_name>\w+)/credentials', DatasetTaskCredentialsHandler, kwargs)
         server.add_route('/healthz', HealthHandler, kwargs)
         server.add_route(r'/(.*)', Error)
 
diff --git a/iceprod/credentials/service.py b/iceprod/credentials/service.py
index bd2906f6..024fd06f 100644
--- a/iceprod/credentials/service.py
+++ b/iceprod/credentials/service.py
@@ -1,47 +1,14 @@
 import asyncio
-import json
 import logging
 import time
 
-from cachetools.func import ttl_cache
 import httpx
 import jwt
-from rest_tools.utils.auth import OpenIDAuth
-
-logger = logging.getLogger('refresh_service')
-
-
-@ttl_cache(maxsize=256, ttl=3600)
-def get_auth(url):
-    return OpenIDAuth(url)
 
+from .util import ClientCreds, get_expiration
 
-def get_expiration(token):
-    """
-    Find a token's expiration time.
-
-    Args:
-        token (str): jwt token
-    Returns:
-        float: expiration unix time
-    """
-    return jwt.decode(token, options={"verify_signature": False})['exp']
-
-
-def is_expired(cred):
-    """
-    Check if an OAuth credential is expired.
-
-    Will mark credential as expired if the access token has less than 5 seconds left.
+logger = logging.getLogger('refresh_service')
 
-    Args:
-        cred (dict): credential dict
-    Returns:
-        bool: True if expired
-    """
-    if cred['type'] != 'oauth':
-        return False
-    return cred['expiration'] < (time.time() + 5)
 
 
 class RefreshService:
@@ -57,7 +24,8 @@ class RefreshService:
     """
     def __init__(self, database, clients, refresh_window, expire_buffer, service_run_interval):
         self.db = database
-        self.clients = json.loads(clients)
+        self.clients = ClientCreds(clients)
+        self.clients.validate()
         self.refresh_window = refresh_window * 3600
         self.expire_buffer = expire_buffer * 3600
 
@@ -71,26 +39,31 @@ async def refresh_cred(self, cred):
             raise Exception('cred does not have a refresh token')
 
         openid_url = jwt.decode(cred['refresh_token'], options={"verify_signature": False})['iss']
-        if openid_url not in self.clients:
+        try:
+            client = self.clients.get_client(openid_url)
+        except KeyError:
             raise Exception('jwt issuer not registered')
-        auth = get_auth(openid_url)
 
         # try the refresh token
         args = {
             'grant_type': 'refresh_token',
             'refresh_token': cred['refresh_token'],
-            'client_id': self.clients[openid_url][0],
+            'client_id': client.client_id,
         }
-        if len(self.clients[openid_url]) > 1:
-            args['client_secret'] = self.clients[openid_url][1]
+        if client.client_secret:
+            args['client_secret'] = client.client_secret
+        if cred.get('scope', None) is not None:
+            args['scope'] = cred['scope']
+
+        logging.warning('refreshing on %s with args %r', client.auth.token_url, args)
 
         new_cred = {}
         try:
-            async with httpx.AsyncClient() as client:
-                r = await client.post(auth.token_url, data=args)
+            async with httpx.AsyncClient() as http_client:
+                r = await http_client.post(client.auth.token_url, data=args)
             r.raise_for_status()
             req = r.json()
-        except httpx.HTTPError as exc:
+        except httpx.HTTPStatusError as exc:
             logger.debug('%r', exc.response.text)
             try:
                 req = exc.response.json()
@@ -103,11 +76,13 @@ async def refresh_cred(self, cred):
             new_cred['access_token'] = req['access_token']
             new_cred['refresh_token'] = req['refresh_token']
             new_cred['expiration'] = get_expiration(req['access_token'])
+            new_cred['scope'] = req.get('scope', cred.get('scope', None))
             logger.debug('%r', new_cred)
 
         return new_cred
 
     def should_refresh(self, cred):
+        logger.info('should_refresh for cred %r', cred)
         now = time.time()
         refresh_exp = now + self.expire_buffer
         last_use_date = now - self.refresh_window
@@ -133,6 +108,7 @@ async def _run_once(self):
                 'type': 'oauth',
                 'last_use': {'$gt': last_use_check},
                 'expiration': {'$lt': exp_check},
+                'scope': {'$exists': True},
             }
 
             user_creds = {}
@@ -145,16 +121,15 @@ async def _run_once(self):
                 if not cred['refresh_token']:
                     logger.info('skipping non-refresh token for user %s, url %s', cred['username'], cred['url'])
                     continue
-                logger.debug('cred: %r', cred)
                 try:
                     if self.should_refresh(cred):
                         args = await self.refresh_cred(cred)
-                        await self.db.user_creds.update_one({'username': cred['username'], 'url': cred['url']}, {'$set': args})
-                        logger.info('refreshed token for user %s, url %s', cred['username'], cred['url'])
+                        await self.db.user_creds.update_one({'username': cred['username'], 'url': cred['url'], 'scope': cred['scope']}, {'$set': args})
+                        logger.info('refreshed token for user %s, url %s, scope %s', cred['username'], cred['url'], cred['scope'])
                     else:
-                        logger.info('not yet time to refresh token for user %s, url %s', cred['username'], cred['url'])
+                        logger.info('not yet time to refresh token for user %s, url %s, scope %s', cred['username'], cred['url'], cred['scope'])
                 except Exception:
-                    logger.error('error refreshing token for user %s, url: %s', cred['username'], cred['url'], exc_info=True)
+                    logger.error('error refreshing token for user %s, url: %s, scope %s', cred['username'], cred['url'], cred['scope'], exc_info=True)
 
             group_creds = {}
             async for row in self.db.group_creds.find(filters, {'_id': False}):
@@ -176,6 +151,28 @@ async def _run_once(self):
                 except Exception:
                     logger.error('error refreshing token for group %s, url: %s', cred['groupname'], cred['url'], exc_info=True)
 
+
+            dataset_creds = []
+            async for row in self.db.dataset_creds.find(filters, {'_id': False}):
+                dataset_creds.append(row)
+
+            for cred in dataset_creds:
+                if cred['type'] != 'oauth':
+                    continue
+                if not cred['refresh_token']:
+                    logger.info('skipping non-refresh token for dataset %s, task, %s, url %s', cred['dataset_id'], cred['task_name'], cred['url'])
+                    continue
+                try:
+                    if self.should_refresh(cred):
+                        args = await self.refresh_cred(cred)
+                        await self.db.dataset_creds.update_one({'dataset_id': cred['dataset_id'], 'task_name': cred['task_name'], 'scope': cred['scope'], 'url': cred['url']}, {'$set': args})
+                        logger.info('refreshed token for dataset %s, task, %s, url %s', cred['dataset_id'], cred['task_name'], cred['url'])
+                    else:
+                        logger.info('not yet time to refresh token for dataset %s, task, %s, url %s', cred['dataset_id'], cred['task_name'], cred['url'])
+                except Exception:
+                    logger.error('error refreshing token for dataset %s, task, %s, url %s', cred['dataset_id'], cred['task_name'], cred['url'], exc_info=True)
+
+
             self.last_success_time = time.time()
         except Exception:
             logger.error('error running refresh', exc_info=True)
diff --git a/iceprod/rest/base_handler.py b/iceprod/rest/base_handler.py
index dd47141f..8f24f89f 100644
--- a/iceprod/rest/base_handler.py
+++ b/iceprod/rest/base_handler.py
@@ -3,6 +3,7 @@
 
 import motor.motor_asyncio
 from rest_tools.server import RestHandlerSetup, RestHandler
+from tornado.escape import json_encode
 
 from iceprod.util import VERSION_STRING
 from iceprod.prom_utils import PromRequestMixin
@@ -50,3 +51,8 @@ def get_current_user(self):
                 logger.info('could not find auth username')
 
         return username
+
+    def write(self, chunk: dict | list) -> None:  # type: ignore[override]
+        """Write dict or list to json"""
+        self.set_header("Content-Type", "application/json; charset=UTF-8")
+        super().write(json_encode(chunk))
diff --git a/iceprod/server/plugins/condor.py b/iceprod/server/plugins/condor.py
index 8fbdba97..27b755ca 100644
--- a/iceprod/server/plugins/condor.py
+++ b/iceprod/server/plugins/condor.py
@@ -19,7 +19,7 @@
 import time
 from typing import Generator, NamedTuple
 
-import htcondor  # type: ignore
+import htcondor2 as htcondor
 from wipac_dev_tools.prometheus_tools import GlobalLabels, AsyncPromWrapper, PromWrapper, AsyncPromTimer, PromTimer
 
 from iceprod.core.config import Task
diff --git a/iceprod/website/data/www_templates/profile.html b/iceprod/website/data/www_templates/profile.html
index 671be8bf..7e3d26b8 100644
--- a/iceprod/website/data/www_templates/profile.html
+++ b/iceprod/website/data/www_templates/profile.html
@@ -63,10 +63,6 @@ <h4>User Credentials</h4>
         <p>TYPE: {{ user_creds[url]["type"]}}</p>
       </div>
     {% end %}
-      <form method="post">
-        {% module xsrf_form_html() %}
-        <button name="add_icecube_token" value="true" type="submit">Add IceCube Token</button>
-      </form>
       <button onclick="$('#add_user_cred_form').show()">Add New User Credentials</button>
       <form method="post" id="add_user_cred_form" style="display: none">
         {% module xsrf_form_html() %}
diff --git a/iceprod/website/server.py b/iceprod/website/server.py
index 9fc1800c..a73ec5a8 100644
--- a/iceprod/website/server.py
+++ b/iceprod/website/server.py
@@ -15,7 +15,7 @@
 import os
 import random
 import re
-from typing import Any
+from typing import Any, Awaitable
 from urllib.parse import urlencode
 
 from iceprod.core.jsonUtil import json_encode
@@ -23,21 +23,19 @@
 from cachetools.func import ttl_cache
 from prometheus_client import Info, start_http_server
 import tornado.web
-import tornado.httpserver
-import tornado.gen
 import jwt
-import tornado.concurrent
+import requests.exceptions
 from rest_tools.client import RestClient, ClientCredentialsAuth
 from rest_tools.server import catch_error, RestServer, RestHandlerSetup, RestHandler, OpenIDCookieHandlerMixin, OpenIDLoginHandler
 from rest_tools.server.session import SessionMixin, Session
 from wipac_dev_tools import from_environment_as_dataclass
 
 from iceprod.util import VERSION_STRING
+from iceprod.credentials.util import ClientCreds
 from iceprod.prom_utils import AsyncMonitor, PromRequestMixin
 from iceprod.roles_groups import GROUPS
 from iceprod.core.config import CONFIG_SCHEMA as DATASET_SCHEMA
 from iceprod.server.config import CONFIG_SCHEMA as SERVER_SCHEMA
-import iceprod.core.functions
 from iceprod.server import documentation
 import iceprod.server.states
 from iceprod.server.util import datetime2str, nowstr
@@ -219,121 +217,182 @@ def clear_tokens(self):
             self._session_mgr.delete_session(username)
 
 
-class TokenStorageMixin(OpenIDCookieHandlerMixin, RestHandler):
+class TokenStorageMixin(RestHandler):
     """
-    Store/load current user's `OpenIDLoginHandler` tokens in iceprod credentials API.
+    Store/load current user's tokens in iceprod credentials API.
     """
-    def initialize(self, cred_rest_client: RestClient, full_url: str, **kwargs):  # type: ignore
+    TokenResult = list[dict[str, Any]]
+
+    def initialize(self, *args, cred_rest_client, **kwargs):
         super().initialize(**kwargs)
         self.cred_rest_client = cred_rest_client
-        self.full_url = full_url
-
-    def get_current_user(self):
-        return None
 
-    async def get_current_user_async(self):
-        """Get the current user, and set auth-related attributes."""
+    async def _get_cred_tokens(self, path: str, url: str | None = None, scope: str | None = None) -> TokenResult:
+        """Get selected tokens from the credential service."""
         try:
             assert self.auth
-            username = self.get_secure_cookie('iceprod_username')
-            if not username:
-                return None
-            if isinstance(username, bytes):
-                username = username.decode('utf-8')
-            creds = await self.cred_rest_client.request('GET', f'/users/{username}/credentials', {'url': self.full_url})
-            cred = creds[self.full_url]
-            access_token = cred['access_token']
-            try:
-                data = self.auth.validate(access_token)
-            except jwt.ExpiredSignatureError:
-                logger.debug('user access_token expired')
-                return None
-            self.auth_data = data
+            args = {'url': url}
+            if scope:
+                args['scope'] = scope
+            creds = await self.cred_rest_client.request('GET', path, args)
+            return creds
+        except requests.exceptions.RequestException:
+            logger.warning('failed to get credentials', exc_info=True)
+        return []
 
-            # lookup groups
-            auth_groups = set()
-            try:
-                for name in GROUPS:
-                    for expression in GROUPS[name]:
-                        ret = eval_expression(data, expression)
-                        auth_groups.update(match.expand(name) for match in ret)
-            except Exception:
-                logger.info('cannot determine groups', exc_info=True)
-            self.auth_groups = sorted(auth_groups)
+    async def get_cred_group_tokens(self, group_name: str, url: str | None = None, scope: str | None = None) -> TokenResult:
+        return await self._get_cred_tokens(
+            path = f'/groups/{group_name}/credentials',
+            url = url, 
+            scope = scope,
+        )
 
-            self.auth_access_token = access_token
-            self.auth_refresh_token = cred.get('refresh_token', '')
-            return username
+    async def get_cred_user_tokens(self, username: str, url: str | None = None, scope: str | None = None) -> TokenResult:
+        return await self._get_cred_tokens(
+            path = f'/users/{username}/credentials',
+            url = url, 
+            scope = scope,
+        )
 
-        except Exception:
-            logger.debug('failed auth', exc_info=True)
-        return None
+    async def get_cred_dataset_tokens(self, dataset_id: str, url: str | None = None, scope: str | None = None) -> TokenResult:
+        return await self._get_cred_tokens(
+            path = f'/datasets/{dataset_id}/credentials',
+            url = url, 
+            scope = scope,
+        )
 
-    def store_tokens(
+    async def get_cred_dataset_task_tokens(self, dataset_id: str, task_name: str, url: str | None = None, scope: str | None = None) -> TokenResult:
+        return await self._get_cred_tokens(
+            path = f'/datasets/{dataset_id}/tasks/{task_name}/credentials',
+            url = url, 
+            scope = scope,
+        )
+
+    async def _put_cred_tokens(
         self,
-        access_token,
-        access_token_exp,
-        refresh_token=None,
-        refresh_token_exp=None,
-        user_info=None,
-        user_info_exp=None,
+        path: str,
+        url: str,
+        access_token: str,
+        refresh_token: str | None =None
     ):
         """
-        Store jwt tokens and user info from OpenID-compliant auth source.
+        Store jwt tokens from OpenID-compliant auth source.
 
         Args:
+            url (str): site url
             access_token (str): jwt access token
-            access_token_exp (int): access token expiration in seconds
             refresh_token (str): jwt refresh token
-            refresh_token_exp (int): refresh token expiration in seconds
-            user_info (dict): user info (from id token or user info lookup)
-            user_info_exp (int): user info expiration in seconds
         """
         assert self.auth
-        if not user_info:
-            user_info = self.auth.validate(access_token)
-        username = user_info.get('preferred_username')
-        if not username:
-            username = user_info.get('upn')
-        if not username:
-            raise tornado.web.HTTPError(400, reason='no username in token')
         args = {
-            'url': self.full_url,
+            'url': url,
             'type': 'oauth',
             'access_token': access_token,
         }
         if refresh_token:
             args['refresh_token'] = refresh_token
 
-        self.cred_rest_client.request_seq('POST', f'/users/{username}/credentials', args)
+        await self.cred_rest_client.request('POST', path, args)
 
-        self.set_secure_cookie('iceprod_username', username, expires_days=30)
+    async def put_cred_group_tokens(
+            self,
+            group_name: str,
+            url: str,
+            access_token: str,
+            refresh_token: str | None = None
+    ):
+        return await self._put_cred_tokens(
+            path = f'/groups/{group_name}/credentials',
+            url = url, 
+            access_token=access_token,
+            refresh_token=refresh_token,
+        )
 
-    def clear_tokens(self):
+    async def put_cred_user_tokens(
+            self,
+            username: str,
+            url: str,
+            access_token: str,
+            refresh_token: str | None = None
+    ):
+        return await self._put_cred_tokens(
+            path = f'/users/{username}/credentials',
+            url = url, 
+            access_token=access_token,
+            refresh_token=refresh_token,
+        )
+
+    async def put_cred_dataset_task_tokens(
+            self,
+            dataset_id: str,
+            task_name: str,
+            url: str,
+            access_token: str,
+            refresh_token: str | None = None
+    ):
+        return await self._put_cred_tokens(
+            path = f'/datasets/{dataset_id}/tasks/{task_name}/credentials',
+            url = url, 
+            access_token=access_token,
+            refresh_token=refresh_token,
+        )
+
+    async def _clear_cred_tokens(self, path: str, url: str | None = None, scope: str | None = None):
         """
-        Clear token data, usually on logout.
+        Clear all token data.
         """
-        self.clear_cookie('iceprod_username')
+        args = {}
+        if url:
+            args['url'] = url
+        if scope:
+            args['scope'] = scope
+        await self.cred_rest_client.request('DELETE', path, args)
+
+    async def clear_cred_group_tokens(self, group_name: str, url: str, scope: str | None = None):
+        return await self._clear_cred_tokens(
+            path = f'/groups/{group_name}/credentials',
+            url = url, 
+            scope = scope,
+        )
+
+    async def clear_cred_user_tokens(self, username: str, url: str, scope: str | None = None):
+        return await self._clear_cred_tokens(
+            path = f'/users/{username}/credentials',
+            url = url, 
+            scope = scope,
+        )
 
+    async def clear_cred_dataset_tokens(self, dataset_id: str, url: str, scope: str | None = None):
+        return await self._clear_cred_tokens(
+            path = f'/datasets/{dataset_id}/credentials',
+            url = url, 
+            scope = scope,
+        )
+
+    async def clear_cred_dataset_task_tokens(self, dataset_id: str, task_name: str, url: str, scope: str | None = None):
+        return await self._clear_cred_tokens(
+            path = f'/datasets/{dataset_id}/tasks/{task_name}/credentials',
+            url = url, 
+            scope = scope,
+        )
 
 class Login(LoginMixin, PromRequestMixin, OpenIDLoginHandler):  # type: ignore
     pass
 
 
-class PublicHandler(LoginMixin, PromRequestMixin, RestHandler):
+class PublicHandler(LoginMixin, TokenStorageMixin, PromRequestMixin, RestHandler):
     """Default Handler"""
-    def initialize(self, rest_api, cred_rest_client, system_rest_client, **kwargs):  # type: ignore
+    def initialize(self, rest_api, system_rest_client, token_clients: list[ClientCreds], **kwargs):  # type: ignore
         """
         Get some params from the website module
 
         :param rest_api: the rest api url
-        :param cred_rest_client: the rest api url for the cred service
         :param system_rest_client: the rest client for the system role
         """
         super().initialize(**kwargs)
         self.rest_api = rest_api
-        self.cred_rest_client = cred_rest_client
         self.system_rest_client = system_rest_client
+        self.token_clients = token_clients
         self.rest_client: RestClient | None = None
 
     def get_template_namespace(self) -> dict[str, Any]:
@@ -691,58 +750,11 @@ async def get(self):
         group_creds = {}
         for g in groups:
             if g != 'users':
-                ret = await self.cred_rest_client.request('GET', f'/groups/{g}/credentials')
-                group_creds[g] = ret
-        user_creds = await self.cred_rest_client.request('GET', f'/users/{username}/credentials')
+                group_creds[g] = await self.get_cred_group_tokens(group_name=g)
+        user_creds = await self.get_cred_user_tokens(username=username)
         self.render('profile.html', username=username, groups=groups,
                     group_creds=group_creds, user_creds=user_creds)
 
-    @authenticated
-    async def post(self):
-        username = self.current_user
-
-        if self.get_argument('add_icecube_token', ''):
-            args = {
-                'url': 'https://data.icecube.aq',
-                'type': 'oauth',
-                'access_token': self.auth_access_token,
-            }
-            if self.auth_refresh_token:
-                args['refresh_token'] = self.auth_refresh_token
-                args['expiration'] = datetime2str(datetime.now(UTC) + timedelta(days=30))
-            await self.cred_rest_client.request('POST', f'/users/{username}/credentials', args)
-
-        else:
-            type_ = self.get_argument('type')
-            args = {
-                'url': self.get_argument('url'),
-                'type': type_,
-            }
-            if type_ == 's3':
-                args['buckets'] = [x for x in self.get_argument('buckets').split('\n') if x]
-                args['access_key'] = self.get_argument('access_key')
-                args['secret_key'] = self.get_argument('secret_key')
-            elif type_ == 'oauth':
-                if acc := self.get_argument('refresh_token', ''):
-                    args['refresh_token'] = acc
-                if ref := self.get_argument('refresh_token', ''):
-                    args['refresh_token'] = ref
-                if not (acc or ref):
-                    raise tornado.web.HTTPError(400, reason='need access or refresh token')
-            else:
-                raise tornado.web.HTTPError(400, reason='bad cred type')
-
-            if self.get_argument('add_user_cred', ''):
-                await self.cred_rest_client.request('POST', f'/users/{username}/credentials', args)
-            elif self.get_argument('add_group_cred', ''):
-                groupname = self.get_argument('group')
-                if groupname not in self.auth_groups or groupname == 'users':
-                    raise tornado.web.HTTPError(400, reason='bad group name')
-                await self.cred_rest_client.request('POST', f'/groups/{groupname}/credentials', args)
-
-        # now show the profile page
-        await self.get()
-
 
 class Logout(PublicHandler):
     @catch_error
@@ -753,6 +765,52 @@ async def get(self):
         self.render('logout.html', status=None)
 
 
+class TokenLogin(TokenStorageMixin, PromRequestMixin, OpenIDLoginHandler):
+    def initialize(self, *args, login_url, oauth_url, **kwargs):
+        super().initialize(*args, **kwargs)
+        self.login_url = login_url
+        self.oauth_url = oauth_url
+
+    def get_login_url(self):
+        return self.login_url
+
+    @catch_error
+    async def get(self):
+        if self.get_argument('error', False):
+            err = self.get_argument('error_description', None)
+            if not err:
+                err = 'unknown oauth2 error'
+            raise tornado.web.HTTPError(400, reason=err)
+        elif self.get_argument('code', False):
+            data = self._decode_state(self.get_argument('state'))
+            dataset_id = data['dataset_id']
+            task_name = data['task_name']
+            tokens = await self.get_authenticated_user(
+                redirect_uri=self.get_login_url(),
+                code=self.get_argument('code'),
+                state=data,
+            )
+            assert self.auth
+            await self.put_cred_dataset_task_tokens(
+                dataset_id=dataset_id,
+                task_name=task_name,
+                url=self.oauth_url,
+                access_token=tokens['access_token'],
+                refresh_token=tokens.get('refresh_token'),
+            )
+            if redirect := data.get('redirect', None):
+                self.redirect(redirect)
+        else:
+            self.oauth_client_scope = ['offline_access']
+            if scope := self.get_argument('scope', None):
+                self.oauth_client_scope.extend(scope.split())
+            state = {
+                'dataset_id': self.get_argument('datset_id'),
+                'task_name': self.get_argument('task_name'),
+            }
+            self.start_oauth_authorization(state)
+
+
 class HealthHandler(PublicHandler):
     """
     Handle health requests.
@@ -800,6 +858,7 @@ class DefaultConfig:
     ICEPROD_CRED_ADDRESS : str = 'https://credentials.iceprod.icecube.aq'
     ICEPROD_CRED_CLIENT_ID : str = ''
     ICEPROD_CRED_CLIENT_SECRET : str = ''
+    TOKEN_CLIENTS : str = '{}'
     REDIS_HOST : str = 'localhost'
     REDIS_USER : str = ''
     REDIS_PASSWORD : str = ''
@@ -869,7 +928,6 @@ def __init__(self):
             raise RuntimeError('ICEPROD_CRED_CLIENT_ID or ICEPROD_CRED_CLIENT_SECRET not specified, and CI_TESTING not enabled!')
 
         handler_args = RestHandlerSetup(rest_config)
-        handler_args['cred_rest_client'] = cred_client
         if config.CI_TESTING:
             self.session = Session()
         else:
@@ -904,9 +962,13 @@ def __init__(self):
         else:
             raise RuntimeError('ICEPROD_API_CLIENT_ID or ICEPROD_API_CLIENT_SECRET not specified, and CI_TESTING not enabled!')
 
+        token_clients = ClientCreds(config.TOKEN_CLIENTS).get_clients()
+
         handler_args.update({
             'rest_api': rest_address,
+            'cred_rest_client': cred_client,
             'system_rest_client': rest_client,
+            'token_clients': token_clients,
         })
         if config.COOKIE_SECRET:
             cookie_secret = config.COOKIE_SECRET
@@ -923,22 +985,34 @@ def __init__(self):
             static_path=static_path,
         )
 
-        server.add_route(r"/", Default, handler_args)
-        server.add_route(r"/submit", Submit, handler_args)
-        server.add_route(r"/config", Config, handler_args)
+        server.add_route("/", Default, handler_args)
+        server.add_route('/submit', Submit, handler_args)
+        server.add_route('/config', Config, handler_args)
         server.add_route(r"/schemas/v3/([\w\.]+)", Schemas, handler_args)
-        server.add_route(r"/dataset", DatasetBrowse, handler_args)
+        server.add_route('/dataset', DatasetBrowse, handler_args)
         server.add_route(r"/dataset/(\w+)", Dataset, handler_args)
         server.add_route(r"/dataset/(\w+)/task", TaskBrowse, handler_args)
         server.add_route(r"/dataset/(\w+)/task/(\w+)", Task, handler_args)
         server.add_route(r"/dataset/(\w+)/job", JobBrowse, handler_args)
         server.add_route(r"/dataset/(\w+)/job/(\w+)", Job, handler_args)
-        server.add_route(r"/help", Help, handler_args)
+        server.add_route('/help', Help, handler_args)
         server.add_route(r"/docs/(.*)", Documentation, handler_args)
         server.add_route(r"/dataset/(\w+)/log/(\w+)", Log, handler_args)
-        server.add_route(r'/profile', Profile, handler_args)
-        server.add_route(r"/login", Login, login_handler_args)
-        server.add_route(r"/logout", Logout, handler_args)
+        server.add_route('/profile', Profile, handler_args)
+        server.add_route('/login', Login, login_handler_args)
+        server.add_route('/logout', Logout, handler_args)
+
+        for client in token_clients:
+            client_handler_args = login_handler_args.copy()
+            client_handler_args['auth'] = client.auth
+            client_handler_args['oauth_client_id'] = client.client_id
+            client_handler_args['oauth_client_secret'] = client.client_secret
+            client_handler_args['oauth_client_scope'] = ''
+            client_handler_args['login_url'] = f'{full_url}/token_clients/{client.id}/login'
+            client_handler_args['oauth_url'] = client.url
+            client_handler_args['cred_rest_client'] = cred_client
+            server.add_route(f'/token_clients/{client.id}/login', TokenLogin, client_handler_args)
+
         server.add_route('/healthz', HealthHandler, handler_args)
         server.add_route(r"/.*", Other, handler_args)
 
diff --git a/pyproject.toml b/pyproject.toml
index 0f859a9c..0ec9ef5b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -12,7 +12,7 @@ dependencies = [
     'cachetools',
     'certifi',
     'cryptography',
-    'htcondor<=24.99.99',
+    'htcondor>=25.3.1',
     'httpx',
     'jsonschema',
     'ldap3',
@@ -29,7 +29,7 @@ dependencies = [
     'tornado',
     'unidecode',
     'wipac-dev-tools',
-    'wipac-rest-tools[redis]>=1.11.6',
+    'wipac-rest-tools[redis]>=1.12.0',
 ]
 dynamic = ["version"] # do not edit — autogenerated by wipac-dev-py-setup-action
 name = "iceprod" # do not edit — autogenerated by wipac-dev-py-setup-action
diff --git a/tests/credentials/test_server.py b/tests/credentials/test_server.py
index f006dc11..0459bc2f 100644
--- a/tests/credentials/test_server.py
+++ b/tests/credentials/test_server.py
@@ -58,13 +58,15 @@ def client(timeout=1, username='user', roles=[], groups=[], exp=10):
 
 GROUP = 'simprod'
 USER = 'username'
+DATASETS = ['dataset1', 'dataset2']
+TASK_NAMES = ['alpha', 'bravo']
 
 
 async def test_credentials_groups_empty(server):
     client = server(roles=['system'])
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
-    assert ret == {}
+    assert ret == []
 
 
 async def test_credentials_groups_s3(server):
@@ -81,7 +83,7 @@ async def test_credentials_groups_s3(server):
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
     data['groupname'] = GROUP
-    assert ret == {data['url']: data}
+    assert ret == [data]
 
     # test bucket in url
     data2 = {
@@ -95,7 +97,7 @@ async def test_credentials_groups_s3(server):
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
     data2['groupname'] = GROUP
-    assert ret == {data['url']: data, data2['url']: data2}
+    assert ret == [data, data2]
 
     # now overwrite
     data3 = {
@@ -110,18 +112,18 @@ async def test_credentials_groups_s3(server):
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
     data3_out = data3.copy()
     data3_out['groupname'] = GROUP
-    assert ret == {data['url']: data3_out, data2['url']: data2}
+    assert ret == [data3_out, data2]
 
     await client.request('DELETE', f'/groups/{GROUP}/credentials', {'url': 'http://foo'})
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
-    assert ret == {data2['url']: data2}
+    assert ret == [data2]
 
     await client.request('POST', f'/groups/{GROUP}/credentials', data3)
     await client.request('DELETE', f'/groups/{GROUP}/credentials')
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
-    assert ret == {}
+    assert ret == []
     
 async def test_credentials_groups_s3_bad(server):
     client = server(roles=['system'])
@@ -180,7 +182,7 @@ async def test_credentials_groups_oauth(server):
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
     for k in data:
-        assert ret[data['url']][k] == data[k]
+        assert ret[0][k] == data[k]
 
 
 async def test_credentials_groups_user(server):
@@ -195,7 +197,7 @@ async def test_credentials_groups_user(server):
 
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
     for k in data:
-        assert ret[data['url']][k] == data[k]
+        assert ret[0][k] == data[k]
 
     data2 = {
         'url': 'http://foo',
@@ -204,13 +206,13 @@ async def test_credentials_groups_user(server):
     }
     await client.request('PATCH', f'/groups/{GROUP}/credentials', data2)
     ret = await client.request('GET', f'/groups/{GROUP}/credentials')
-    assert ret[data2['url']]['refresh_token'] == data2['refresh_token']
+    assert ret[0]['refresh_token'] == data2['refresh_token']
 
     await client.request('DELETE', f'/groups/{GROUP}/credentials')
 
     # bad user
     client = server(roles=['user'], groups=['users'])
-    
+
     with pytest.raises(requests.exceptions.HTTPError) as exc_info:
         await client.request('POST', f'/groups/{GROUP}/credentials', data)
     assert exc_info.value.response.status_code == 403
@@ -228,7 +230,7 @@ async def test_credentials_users_empty(server):
     client = server(roles=['system'])
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
-    assert ret == {}
+    assert ret == []
 
 
 async def test_credentials_users_s3(server):
@@ -245,7 +247,7 @@ async def test_credentials_users_s3(server):
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
     data['username'] = USER
-    assert ret == {data['url']: data}
+    assert ret == [data]
 
     data2 = {
         'url': 'http://bar',
@@ -258,7 +260,7 @@ async def test_credentials_users_s3(server):
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
     data2['username'] = USER
-    assert ret == {data['url']: data, data2['url']: data2}
+    assert ret == [data, data2]
 
     # now overwrite
     data3 = {
@@ -273,18 +275,18 @@ async def test_credentials_users_s3(server):
     ret = await client.request('GET', f'/users/{USER}/credentials')
     data3_out = data3.copy()
     data3_out['username'] = USER
-    assert ret == {data['url']: data3_out, data2['url']: data2}
+    assert ret == [data3_out, data2]
 
     await client.request('DELETE', f'/users/{USER}/credentials', {'url': 'http://foo'})
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
-    assert ret == {data2['url']: data2}
+    assert ret == [data2]
 
     await client.request('POST', f'/users/{USER}/credentials', data3)
     await client.request('DELETE', f'/users/{USER}/credentials')
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
-    assert ret == {}
+    assert ret == []
 
 
 async def test_credentials_users_oauth(server):
@@ -299,7 +301,7 @@ async def test_credentials_users_oauth(server):
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
     for k in data:
-        assert ret[data['url']][k] == data[k]
+        assert ret[0][k] == data[k]
 
 
 async def test_credentials_users_user(server):
@@ -314,7 +316,7 @@ async def test_credentials_users_user(server):
 
     ret = await client.request('GET', f'/users/{USER}/credentials')
     for k in data:
-        assert ret[data['url']][k] == data[k]
+        assert ret[0][k] == data[k]
 
     data2 = {
         'url': 'http://foo',
@@ -323,7 +325,7 @@ async def test_credentials_users_user(server):
     }
     await client.request('PATCH', f'/users/{USER}/credentials', data2)
     ret = await client.request('GET', f'/users/{USER}/credentials', {'norefresh': 'true'})
-    assert ret[data2['url']]['last_use'] == data2['last_use']
+    assert ret[0]['last_use'] == data2['last_use']
 
     await client.request('DELETE', f'/users/{USER}/credentials')
 
@@ -343,6 +345,192 @@ async def test_credentials_users_user(server):
     assert exc_info.value.response.status_code == 403
 
 
+async def test_credentials_datasets_user_fail(server):
+    client = server(roles=['user'], groups=['users', GROUP])
+    
+    data = {
+        'url': 'http://foo',
+        'type': 'oauth',
+    }
+    with pytest.raises(requests.exceptions.HTTPError) as exc_info:
+        await client.request('GET', f'/datasets/{DATASETS[0]}/credentials', data)
+    assert exc_info.value.response.status_code == 403
+
+    data = {
+        'url': 'http://foo',
+        'type': 'oauth',
+        'access_token': str(client.access_token),
+    }
+    with pytest.raises(requests.exceptions.HTTPError) as exc_info:
+        await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data)
+    assert exc_info.value.response.status_code == 403
+
+
+async def test_credentials_datasets(server):
+    client = server(roles=['system'])
+    auth = Auth('secret')
+
+    data = {
+        'url': 'http://foo',
+        'type': 'oauth',
+        'access_token': str(client.access_token),
+    }
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    for k in data:
+        assert ret[0][k] == data[k]
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/credentials')
+    for k in data:
+        assert ret[0][k] == data[k]
+
+    data2 = {
+        'url': 'http://foo',
+        'access_token': str(client.access_token),
+        'last_use': time.time()
+    }
+    await client.request('PATCH', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data2)
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', {'norefresh': 'true'})
+    assert ret[0]['last_use'] == data2['last_use']
+
+    # now for the second task
+    data3 = {
+        'url': 'http://foo',
+        'type': 'oauth',
+        'access_token': auth.create_token('sub2'),
+    }
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[1]}/credentials', data3)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    assert len(ret) == 1
+    assert ret[0]['access_token'] == data2['access_token']
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[1]}/credentials')
+    assert len(ret) == 1
+    assert ret[0]['access_token'] == data3['access_token']
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/credentials')
+    assert len(ret) == 2
+    assert ret[0]['access_token'] == data2['access_token']
+    assert ret[1]['access_token'] == data3['access_token']
+
+    # now for a different dataset
+    data4 = {
+        'url': 'http://foo',
+        'type': 'oauth',
+        'access_token': auth.create_token('sub3'),
+    }
+    await client.request('POST', f'/datasets/{DATASETS[1]}/tasks/{TASK_NAMES[1]}/credentials', data4)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[1]}/tasks/{TASK_NAMES[0]}/credentials')
+    assert len(ret) == 0
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[1]}/tasks/{TASK_NAMES[1]}/credentials')
+    assert len(ret) == 1
+    assert ret[0]['access_token'] == data4['access_token']
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[1]}/credentials')
+    assert len(ret) == 1
+    assert ret[0]['access_token'] == data4['access_token']
+    
+    # delete
+    await client.request('DELETE', f'/datasets/{DATASETS[0]}/credentials')
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/credentials')
+    assert len(ret) == 0
+    
+    ret = await client.request('GET', f'/datasets/{DATASETS[1]}/credentials')
+    assert len(ret) == 1
+    assert ret[0]['access_token'] == data4['access_token']
+
+    await client.request('DELETE', f'/datasets/{DATASETS[1]}/credentials')
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/credentials')
+    assert len(ret) == 0
+    ret = await client.request('GET', f'/datasets/{DATASETS[1]}/credentials')
+    assert len(ret) == 0
+
+
+async def test_credentials_datasets_empty(server):
+    client = server(roles=['system'])
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/credentials')
+    assert ret == []
+
+
+async def test_credentials_datasets_s3(server):
+    client = server(roles=['system'])
+
+    data = {
+        'url': 'http://foo',
+        'type': 's3',
+        'access_key': 'XXXX',
+        'secret_key': 'YYYY',
+        'buckets': ['bar'],
+    }
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/credentials')
+    data['dataset_id'] = DATASETS[0]
+    data['task_name'] = TASK_NAMES[0]
+    assert ret == [data]
+
+    data2 = {
+        'url': 'http://bar',
+        'type': 's3',
+        'access_key': 'XXXX',
+        'secret_key': 'YYYY',
+        'buckets': ['bar'],
+    }
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data2)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    data2['dataset_id'] = DATASETS[0]
+    data2['task_name'] = TASK_NAMES[0]
+    assert ret == [data, data2]
+
+    # now overwrite
+    data3 = {
+        'url': 'http://foo',
+        'type': 's3',
+        'access_key': 'XXXX',
+        'secret_key': 'YYYY',
+        'buckets': ['baz'],
+    }
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data3)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    data3_out = data3.copy()
+    data3_out['dataset_id'] = DATASETS[0]
+    data3_out['task_name'] = TASK_NAMES[0]
+    assert ret == [data3_out, data2]
+
+    await client.request('DELETE', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', {'url': 'http://foo'})
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    assert ret == [data2]
+
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data3)
+    await client.request('DELETE', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    assert ret == []
+
+
+async def test_credentials_datasets_oauth(server):
+    client = server(roles=['system'])
+
+    data = {
+        'url': 'http://foo',
+        'type': 'oauth',
+        'access_token': str(client.access_token),
+    }
+    await client.request('POST', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials', data)
+
+    ret = await client.request('GET', f'/datasets/{DATASETS[0]}/tasks/{TASK_NAMES[0]}/credentials')
+    for k in data:
+        assert ret[0][k] == data[k]
+
+
 async def test_credentials_healthz(server):
     client = server(username=USER, roles=['user'], groups=['users'])
 
diff --git a/tests/credentials/test_service.py b/tests/credentials/test_service.py
index 9f971180..2eb374ae 100644
--- a/tests/credentials/test_service.py
+++ b/tests/credentials/test_service.py
@@ -6,31 +6,48 @@
 import pytest
 
 import iceprod.credentials.service
+import iceprod.credentials.util
 
 
 GROUP = 'simprod'
 USER = 'username'
+DATASETS = ['dataset1', 'dataset2']
+TASK_NAMES = ['alpha', 'bravo']
 
 
-def test_credentials_service_get_expiration():
-    t = time.time()
-    tok = jwt.encode({'exp': t}, 'secret')
-    e = iceprod.credentials.service.get_expiration(tok)
-    assert t == e
+@pytest.mark.respx(using="httpx")
+async def test_credentials_service_refresh_cred(respx_mock, monkeypatch):
+    clients = json.dumps({
+        'http://iceprod.test': {'client_id': 'id', 'client_secret': 'secret'}
+    })
+    rs = iceprod.credentials.service.RefreshService(None, clients, 1, 1, 60)
 
+    now = time.time()
 
-def test_credentials_service_is_expired():
     cred = {
+        'url': 'http://iceprod.test',
         'type': 'oauth',
-        'expiration': time.time()
+        'expiration': now,
+        'last_use': now,
+        'refresh_token': jwt.encode({'exp': now, 'iss': 'http://iceprod.test'}, 'secret'),
     }
-    assert iceprod.credentials.service.is_expired(cred)
 
-    cred = {
-        'type': 'oauth',
-        'expiration': time.time()+100
-    }
-    assert not iceprod.credentials.service.is_expired(cred)
+    ref = jwt.encode({'exp': now+1000, 'iss': 'http://iceprod.test', 'refresh': 'refresh'}, 'secret')
+    mock = respx_mock.post("http://iceprod.test/").respond(200, json={
+        'access_token': jwt.encode({'exp': now+1000, 'iss': 'http://iceprod.test'}, 'secret'),
+        'refresh_token': ref
+    })
+
+    authmock = MagicMock()
+    authmock.return_value.token_url = 'http://iceprod.test'
+    monkeypatch.setattr(iceprod.credentials.util, '_get_auth', authmock)
+
+    new_cred = await rs.refresh_cred(cred)
+
+    assert mock.called
+
+    assert new_cred['refresh_token'] == ref
+    assert new_cred['expiration'] == now+1000
 
 
 async def test_credentials_service_refresh_empty(mongo_url, mongo_clear, respx_mock):
@@ -72,7 +89,7 @@ async def test_credentials_service_refresh_not_exp(mongo_url, mongo_clear, respx
 async def test_credentials_service_refresh_group(mongo_url, mongo_clear, respx_mock, monkeypatch):
     db = motor.motor_asyncio.AsyncIOMotorClient(mongo_url)['creds']
     clients = json.dumps({
-        'http://iceprod.test': ['client-id', 'client-secret']
+        'http://iceprod.test': {'client_id': 'id', 'client_secret': 'secret'}
     })
     rs = iceprod.credentials.service.RefreshService(db, clients, 1, 1, 60)
 
@@ -85,6 +102,7 @@ async def test_credentials_service_refresh_group(mongo_url, mongo_clear, respx_m
         'expiration': now,
         'last_use': now,
         'refresh_token': jwt.encode({'exp': now, 'iss': 'http://iceprod.test'}, 'secret'),
+        'scope': '',
     })
 
     ref = jwt.encode({'exp': now+1000, 'iss': 'http://iceprod.test', 'refresh': 'refresh'}, 'secret')
@@ -95,13 +113,14 @@ async def test_credentials_service_refresh_group(mongo_url, mongo_clear, respx_m
 
     authmock = MagicMock()
     authmock.return_value.token_url = 'http://iceprod.test'
-    monkeypatch.setattr(iceprod.credentials.service, 'get_auth', authmock)
+    monkeypatch.setattr(iceprod.credentials.util, '_get_auth', authmock)
 
     await rs._run_once()
 
     assert mock.called
 
     ret = await db.group_creds.find_one({'url': 'http://iceprod.test'})
+    assert ret
     assert ret['refresh_token'] == ref
     assert ret['expiration'] == now+1000
 
@@ -110,7 +129,7 @@ async def test_credentials_service_refresh_group(mongo_url, mongo_clear, respx_m
 async def test_credentials_service_refresh_user(mongo_url, mongo_clear, respx_mock, monkeypatch):
     db = motor.motor_asyncio.AsyncIOMotorClient(mongo_url)['creds']
     clients = json.dumps({
-        'http://iceprod.test': ['client-id', 'client-secret']
+        'http://iceprod.test': {'client_id': 'id', 'client_secret': 'secret'}
     })
     rs = iceprod.credentials.service.RefreshService(db, clients, 1, 1, 60)
 
@@ -123,6 +142,7 @@ async def test_credentials_service_refresh_user(mongo_url, mongo_clear, respx_mo
         'expiration': now,
         'last_use': now,
         'refresh_token': jwt.encode({'exp': now, 'iss': 'http://iceprod.test'}, 'secret'),
+        'scope': '',
     })
 
     ref = jwt.encode({'exp': now+1000, 'iss': 'http://iceprod.test', 'refresh': 'refresh'}, 'secret')
@@ -133,12 +153,56 @@ async def test_credentials_service_refresh_user(mongo_url, mongo_clear, respx_mo
 
     authmock = MagicMock()
     authmock.return_value.token_url = 'http://iceprod.test'
-    monkeypatch.setattr(iceprod.credentials.service, 'get_auth', authmock)
+    monkeypatch.setattr(iceprod.credentials.util, '_get_auth', authmock)
 
     await rs._run_once()
 
+    assert authmock.called
     assert mock.called
 
     ret = await db.user_creds.find_one({'url': 'http://iceprod.test'})
+    assert ret
+    assert ret['refresh_token'] == ref
+    assert ret['expiration'] == now+1000
+
+
+@pytest.mark.respx(using="httpx")
+async def test_credentials_service_refresh_dataset(mongo_url, mongo_clear, respx_mock, monkeypatch):
+    db = motor.motor_asyncio.AsyncIOMotorClient(mongo_url)['creds']
+    clients = json.dumps({
+        'http://iceprod.test': {'client_id': 'id', 'client_secret': 'secret'}
+    })
+    rs = iceprod.credentials.service.RefreshService(db, clients, 1, 1, 60)
+
+    now = time.time()
+
+    await db.dataset_creds.insert_one({
+        'dataset_id': DATASETS[0],
+        'task_name': TASK_NAMES[0],
+        'url': 'http://iceprod.test',
+        'type': 'oauth',
+        'expiration': now,
+        'last_use': now,
+        'refresh_token': jwt.encode({'exp': now, 'iss': 'http://iceprod.test'}, 'secret'),
+        'scope': '',
+    })
+
+    ref = jwt.encode({'exp': now+1000, 'iss': 'http://iceprod.test', 'refresh': 'refresh'}, 'secret')
+    mock = respx_mock.post("http://iceprod.test/").respond(200, json={
+        'access_token': jwt.encode({'exp': now+1000, 'iss': 'http://iceprod.test'}, 'secret'),
+        'refresh_token': ref
+    })
+
+    authmock = MagicMock()
+    authmock.return_value.token_url = 'http://iceprod.test'
+    monkeypatch.setattr(iceprod.credentials.util, '_get_auth', authmock)
+
+    await rs._run_once()
+
+    assert authmock.called
+    assert mock.called
+
+    ret = await db.dataset_creds.find_one({'url': 'http://iceprod.test'})
+    assert ret
     assert ret['refresh_token'] == ref
     assert ret['expiration'] == now+1000
diff --git a/tests/server/plugins/condor_test.py b/tests/server/plugins/condor_test.py
index bd7546e4..26780533 100644
--- a/tests/server/plugins/condor_test.py
+++ b/tests/server/plugins/condor_test.py
@@ -6,7 +6,7 @@
 import time
 from unittest.mock import MagicMock, AsyncMock
 
-import htcondor
+import htcondor2 as htcondor
 import pytest
 
 from iceprod.core.config import Job, Task