Repository: astropy/pyvo
Pull request #157 (2019-06-28 22:20:42)

Auth
Sorry for making a different branch and review, but I am trying to keep things clean and have the ability to look at what I've done in the other branch.

I've changed a lot of things around based on feedback:

- Session is now the thing that is associated with a securitymethod URI, since in requests you can really do anything with a session.
- Session is passed into each object via keyword arg, and creates one if it doesn't exist.
- AuthSession now ducktypes for pyvo as a session, since pyvo only calls get and post.  By not having it subclass session, properties that could alter behavior have no effect.
- Helper methods for easy session configuration for cookies, basic auth, and certificates were added.
- AuthSession is passed in to attach it to a service.  It does this by looking for a special parse_capabilities attribute.
- I've moved the 'tests', which were really more examples, to the examples directory.  

If people think this is looking close, I will start writing unit tests against it that don't require a server to keep it in line with the other pyvo tests.  Also, updating doc strings, etc.

cbanek: Okay so here I will discuss my reasoning to use an object over a dict, and why I think this is the right design decision:

First off, auth is complicated.  It's error prone, it's complicated, it involves secrets and possibly exposing secrets.  So we have to be careful.  Things need to be very obvious and explicit.  With this in mind...

A dict has some nice properties.  It's an easy to work with data type, and it can contain pretty much anything.  But in the end, it is a data type, and holds data - not logic.  If I were passing in a dict as you say, I would still need all the layers underneath it to actually do the logic.  So this doesn't really change the underlying implementation, but only the way it looks to the caller.

(Note: Again, just the creds isn't enough, we also need the capabilities of the service, and all this data needs to be passed down to child objects like TAPResults, etc. since they don't have access to the capabilities of the service who created them otherwise)

Now, is allowing the user to pass in a dict a good user interface?  I would argue that an object is a better interface.  In the object I have created, I have functions that people can look up (eventually with docstrings) that will say what functions do what, and the functions are very simply named.  This would be opposed to the schema of the dict, which could be complicated, and harder to document.  Given that python is very type loose, understanding this schema is everything.  For example, in this PR you said that we could pass in SSO_COOKIE both as a cookiejar and a dict.  I think it's less cognitive load to just call a function and pass in either the cookiejar or the cookie, and I can validate that what is passed in is the expected type.  If not, we could throw an exception right there.

The data model of this schema is also likely to be complex and need to be changed, at the very least to add new keys to support new auth methods, and plumbing that down through.  With the PR currently as posted, someone could write a new auth method and attach it to an auth URI and have it work with no changes to pyvo.  I think that's great.

This may be a bit of my PTSD from Kubernetes where it takes yaml, and you need to discover all the keys, and it will sometimes silently do bad things for you, but I find an actual interface with functions and objects to be much preferable to trying to shove everything into a data model where what each key does is less clear.  Many times to understand what I can pass in to some of these dicts or kwargs, I need to look deeply into the code.  I really don't want that to be the case with auth.
cbanek: Okay, lots of changes:
1. Tests
2. Docstrings updated
3. Add cookie_jar setting in credstore
4. Fix test problem with loading pkg resources.

Build is green.
cbanek: Okay, some new changes.

1. Change parse_capabilities to update_from_capabilities (just a naming change)
2. I found some issues the other day with how hitting the capabilities and availability endpoints from the mixins weren't using the sessions and were using requests directly.  So I made these calls use the same session variable on the class. (also will help with things like user-agent)
3. I've also added a function to authsession to wrap authurls, where someone can add a URL and associate it with a security method outside of capabilities.  While the capabilities is nice to do the negotiation, sometimes you want to set it for services that don't advertise, or don't have capabilities, but you know you want to use a particular auth method.  This provides a good catch all to force negotiation.
timj: I do prefer a class to a dict since it does give you more control.

@andamian it looks to me like @cbanek has dealt with your review comments. Can you please respond and indicate whether you approve the current version or feel there is a fundamental problem with the PR. We would like to get unblocked on this but the PR requires at least one approval from someone.

 @bmerin or @jcsegovia do you have any intention of reviewing?  Auth is really important for the integration of ESAC into pyvo so it would be great if you could sign off.
andamian: > I do prefer a class to a dict since it does give you more control.
> 
> @andamian it looks to me like @cbanek has dealt with your review comments. Can you please respond and indicate whether you approve the current version or feel there is a fundamental problem with the PR. We would like to get unblocked on this but the PR requires at least one approval from someone.
> 
> @bmerin or @jcsegovia do you have any intention of reviewing? Auth is really important for the integration of ESAC into pyvo so it would be great if you could sign off.

Sorry - I was busy/away in vacation. I'm in the process of reviewing the latest changes. At one point I think @msdemlei also expressed an interest in having a look. I agree that we should get this closed - I'd like to start using it before the release of 1.0.
cbanek: Okay pushed a new commit to get out some of the string constants for URIs for different security methods.  Hopefully we can now call this done.

Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index df8539b..93b7a27 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -7,6 +7,7 @@
 
 - Sphinx 1.7 or higher is needed to build the documentation. [#160]
 
+- Add support for authenticated requests. [#157]
 
 0.9.3 (2019-05-30)
 ==================
diff --git a/examples/auth/cadc_auth_example.py b/examples/auth/cadc_auth_example.py
new file mode 100644
index 0000000..1548912
--- /dev/null
+++ b/examples/auth/cadc_auth_example.py
@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+"""
+Example for authenticating with CADC TAP service
+"""
+import getpass
+
+import requests
+
+import pyvo
+import pyvo.extensions.auth.authsession
+
+# Gather login information
+data = {
+    'username': input('Username:'),
+    'password': getpass.getpass('Password:')
+}
+
+headers = {
+    'Content-Type': 'application/x-www-form-urlencoded',
+    'Accept': 'text/plain'
+}
+
+# Do the login and get the cookie
+login_url = 'https://ws-cadc.canfar.net/ac/login'
+response = requests.post(login_url, data=data, headers=headers)
+response.raise_for_status()
+cookie = '\"' + response.text + '\"'
+
+# Configure the session and run the query
+auth = pyvo.extensions.auth.authsession.AuthSession()
+auth.credentials.set_cookie('CADC_SSO', cookie)
+service = pyvo.dal.TAPService('https://www.cadc-ccda.hia-iha.nrc-cnrc.gc.ca/tap', auth)
+job = service.search('SELECT * from TAP_SCHEMA.tables')
+print(job)
diff --git a/examples/auth/cert_auth_example.py b/examples/auth/cert_auth_example.py
new file mode 100644
index 0000000..1a0d136
--- /dev/null
+++ b/examples/auth/cert_auth_example.py
@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+"""
+Example for using client certificate authentication with CADC TAP service
+Tests for pyvo.extensions.authsession
+"""
+import pyvo
+import pyvo.extensions.auth.authsession
+
+certificate_path = input('Path to client certificate file:')
+auth = pyvo.extensions.auth.authsession.AuthSession()
+auth.credentials.set_client_certificate(certificate_path)
+service = pyvo.dal.TAPService('https://www.cadc-ccda.hia-iha.nrc-cnrc.gc.ca/tap', auth)
+job = service.search('SELECT * from TAP_SCHEMA.tables')
+print(job)
diff --git a/examples/auth/gaia_auth_example.py b/examples/auth/gaia_auth_example.py
new file mode 100644
index 0000000..49baab1
--- /dev/null
+++ b/examples/auth/gaia_auth_example.py
@@ -0,0 +1,37 @@
+#!/usr/bin/env python
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+"""
+Example for authenticating with Gaia TAP service
+"""
+import getpass
+
+import requests
+
+import pyvo
+import pyvo.extensions.auth.authsession
+from pyvo.extensions.auth.securitymethods import ANONYMOUS
+
+# Gather login information
+data = {
+    'username': input('Username:'),
+    'password': getpass.getpass('Password:')
+}
+
+headers = {
+    'Content-Type': 'application/x-www-form-urlencoded',
+    'Accept': 'text/plain'
+}
+
+# Create a session and do the login.
+# The cookie will end up in the cookie jar of the session.
+login_url = 'http://gea.esac.esa.int/tap-server/login'
+session = requests.Session()
+response = session.post(login_url, data=data, headers=headers)
+response.raise_for_status()
+
+# Use this session with the auth cookie for all requests to Gaia.
+auth = pyvo.extensions.auth.authsession.AuthSession()
+auth.credentials.set(ANONYMOUS, session)
+service = pyvo.dal.TAPService('http://gea.esac.esa.int/tap-server/tap', auth)
+job = service.search('SELECT * from TAP_SCHEMA.tables')
+print(job)
diff --git a/pyvo/dal/adhoc.py b/pyvo/dal/adhoc.py
index 05ce484..8ab69de 100644
--- a/pyvo/dal/adhoc.py
+++ b/pyvo/dal/adhoc.py
@@ -17,7 +17,6 @@ from astropy.io.votable.tree import Resource, Group
 from astropy.utils.collections import HomogeneousList
 
 from ..utils.decorators import stream_decode_content
-from ..utils.http import session
 
 
 # monkeypatch astropy with group support in RESOURCE
@@ -89,8 +88,8 @@ class AdhocServiceResultsMixin:
     """
     Mixing for adhoc:service functionallity for results classes.
     """
-    def __init__(self, votable, url=None):
-        super().__init__(votable, url=url)
+    def __init__(self, votable, url=None, session=None):
+        super().__init__(votable, url=url, session=session)
 
         self._adhocservices = list(
             resource for resource in votable.resources
@@ -178,7 +177,7 @@ class DatalinkRecordMixin:
     def getdataset(self, timeout=None):
         try:
             url = next(self.getdatalink().bysemantics('#this')).access_url
-            response = session.get(url, stream=True, timeout=timeout)
+            response = self._session.get(url, stream=True, timeout=timeout)
             response.raise_for_status()
             return response.raw
         except (DALServiceError, ValueError, StopIteration):
@@ -191,7 +190,7 @@ class DatalinkService(DALService, AvailabilityMixin, CapabilityMixin):
     a representation of a Datalink service
     """
 
-    def __init__(self, baseurl):
+    def __init__(self, baseurl, session=None):
         """
         instantiate a Datalink service
 
@@ -199,8 +198,17 @@ class DatalinkService(DALService, AvailabilityMixin, CapabilityMixin):
         ----------
         baseurl :  str
            the base URL that should be used for forming queries to the service.
+        session : object
+           optional session to use for network requests
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
+
+        # Check if the session has an update_from_capabilities attribute.
+        # This means that the session is aware of IVOA capabilities,
+        # and can use this information in processing network requests.
+        # One such usecase for this is auth.
+        if hasattr(self._session, 'update_from_capabilities'):
+            self._session.update_from_capabilities(self.capabilities)
 
     def run_sync(self, id, responseformat=None, **keywords):
         """
@@ -258,6 +266,9 @@ class DatalinkQuery(DALQuery):
     :py:attr:`~pyvo.dal.query.DALQuery.baseurl` to send a configured
     query to another service.
 
+    A session can also optionally be passed in that will be used for
+    network transactions made by this object to remote services.
+
     In addition to the search constraint attributes described below, search
     parameters can be set generically by name via dict semantics.
     The typical function for submitting the query is ``execute()``; however,
@@ -265,7 +276,7 @@ class DatalinkQuery(DALQuery):
     allowing the caller to take greater control of the result processing.
     """
     @classmethod
-    def from_resource(cls, row, resource, **kwargs):
+    def from_resource(cls, row, resource, session=None, **kwargs):
         """
         Creates a instance from a Record and a Datalink Resource.
 
@@ -313,10 +324,10 @@ class DatalinkQuery(DALQuery):
             except KeyError:
                 query_params[name] = query_param
 
-        return cls(accessurl, **query_params)
+        return cls(accessurl, session=session, **query_params)
 
     def __init__(
-            self, baseurl, id=None, responseformat=None, **keywords):
+            self, baseurl, id=None, responseformat=None, session=None, **keywords):
         """
         initialize the query object with the given parameters
 
@@ -328,8 +339,10 @@ class DatalinkQuery(DALQuery):
             the dataset identifier
         responseformat : str
             the output format
+        session : object
+            optional session to use for network requests
         """
-        super().__init__(baseurl, **keywords)
+        super().__init__(baseurl, session=session, **keywords)
 
         if id:
             self["ID"] = id
@@ -350,7 +363,7 @@ class DatalinkQuery(DALQuery):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return DatalinkResults(self.execute_votable(), url=self.queryurl)
+        return DatalinkResults(self.execute_votable(), url=self.queryurl, session=self._session)
 
 
 class DatalinkResults(DatalinkResultsMixin, DALResults):
@@ -424,7 +437,7 @@ class DatalinkResults(DatalinkResultsMixin, DALResults):
         --------
         Record
         """
-        return DatalinkRecord(self, index)
+        return DatalinkRecord(self, index, session=self._session)
 
     def bysemantics(self, semantics):
         """
diff --git a/pyvo/dal/mimetype.py b/pyvo/dal/mimetype.py
index b8ef987..246a08c 100644
--- a/pyvo/dal/mimetype.py
+++ b/pyvo/dal/mimetype.py
@@ -8,7 +8,7 @@ import mimeparse
 
 from astropy.io.fits import HDUList
 
-from ..utils.http import session
+from ..utils.http import use_session
 
 
 mimetypes.add_type('application/fits', 'fits')
@@ -56,7 +56,7 @@ def mime2extension(mimetype, default=None):
     return ext
 
 
-def mime_object_maker(url, mimetype):
+def mime_object_maker(url, mimetype, session=None):
     """
     return a data object suitable for the mimetype given.
     this will either return a astropy fits object or a pyvo DALResults object,
@@ -68,7 +68,10 @@ def mime_object_maker(url, mimetype):
         the object download url
     mimetype : str
         the content mimetype
+    session : object
+        optional session to use for network requests
     """
+    session = use_session(session)
     mimetype = mimeparse.parse_mime_type(mimetype)
 
     if mimetype[0] == 'text':
diff --git a/pyvo/dal/query.py b/pyvo/dal/query.py
index 985f51a..0ce777d 100644
--- a/pyvo/dal/query.py
+++ b/pyvo/dal/query.py
@@ -45,7 +45,7 @@ from .exceptions import (DALFormatError, DALServiceError, DALQueryError)
 from .. import samp
 
 from ..utils.decorators import stream_decode_content
-from ..utils.http import session
+from ..utils.http import use_session
 
 
 class DALService:
@@ -54,7 +54,7 @@ class DALService:
     endpoint.
     """
 
-    def __init__(self, baseurl):
+    def __init__(self, baseurl, session=None):
         """
         instantiate the service connecting it to a base URL
 
@@ -62,8 +62,11 @@ class DALService:
         ----------
         baseurl :  str
            the base URL that should be used for forming queries to the service.
+        session : object
+           optional session to use for network requests
         """
         self._baseurl = baseurl
+        self._session = use_session(session)
 
     @property
     def baseurl(self):
@@ -105,7 +108,7 @@ class DALService:
         DALQuery
            a generic query object
         """
-        q = DALQuery(self.baseurl, **keywords)
+        q = DALQuery(self.baseurl, session=self._session, **keywords)
         return q
 
     def describe(self):
@@ -119,11 +122,14 @@ class DALQuery(dict):
     functions will submit the query and return the results.
 
     The base URL for the query can be changed via the baseurl property.
+
+    A session can also optionally be passed in that will be used for
+    network transactions made by this object to remote services.
     """
 
     _ex = None
 
-    def __init__(self, baseurl, **keywords):
+    def __init__(self, baseurl, session=None, **keywords):
         """
         initialize the query object with a baseurl
         """
@@ -131,6 +137,7 @@ class DALQuery(dict):
             baseurl = baseurl.decode("utf-8")
 
         self._baseurl = baseurl.rstrip("?")
+        self._session = use_session(session)
 
         self.update({key.upper(): value for key, value in keywords.items()})
 
@@ -156,7 +163,7 @@ class DALQuery(dict):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return DALResults(self.execute_votable(), self.queryurl)
+        return DALResults(self.execute_votable(), self.queryurl, session=self._session)
 
     def execute_raw(self):
         """
@@ -198,7 +205,7 @@ class DALQuery(dict):
         url = self.queryurl
         params = {k: v for k, v in self.items()}
 
-        response = session.get(url, params=params, stream=True)
+        response = self._session.get(url, params=params, stream=True)
         return response
 
     def execute_votable(self):
@@ -257,19 +264,23 @@ class DALResults:
     """
     @classmethod
     @stream_decode_content
-    def _from_result_url(cls, result_url):
+    def _from_result_url(cls, result_url, session):
         return session.get(result_url, stream=True).raw
 
     @classmethod
-    def from_result_url(cls, result_url):
+    def from_result_url(cls, result_url, session=None):
         """
         Create a result object from a url.
+
+        Uses the optional session to make the request.
         """
+        session = use_session(session)
         return cls(
-            votableparse(cls._from_result_url(result_url).read),
-            url=result_url)
+            votableparse(cls._from_result_url(result_url, session).read),
+            url=result_url,
+            session=session)
 
-    def __init__(self, votable, url=None):
+    def __init__(self, votable, url=None, session=None):
         """
         initialize the cursor.  This constructor is not typically called
         by directly applications; rather an instance is obtained from calling
@@ -282,6 +293,8 @@ class DALResults:
            astropy.io.votable.tree.VOTableFile instance.
         url : str
            the URL that produced the response
+        session : object
+           optional session to use for network requests
 
         Raises
         ------
@@ -295,6 +308,8 @@ class DALResults:
         self._votable = votable
 
         self._url = url
+        self._session = use_session(session)
+
         self._status = self._findstatus(votable)
         if self._status[0].lower() not in ("ok", "overflow"):
             raise DALQueryError(self._status[1], self._status[0], url)
@@ -525,7 +540,7 @@ class DALResults:
         --------
         Record
         """
-        return Record(self, index)
+        return Record(self, index, session=self._session)
 
     def getvalue(self, name, index):
         """
@@ -611,9 +626,10 @@ class Record(Mapping):
     additional functions for access to service type-specific data.
     """
 
-    def __init__(self, results, index):
+    def __init__(self, results, index, session=None):
         self._results = results
         self._index = index
+        self._session = use_session(session)
         self._mapping = collections.OrderedDict(
             zip(
                 results.fieldnames,
@@ -738,9 +754,9 @@ class Record(Mapping):
             raise KeyError("no dataset access URL recognized in record")
 
         if timeout:
-            response = session.get(url, stream=True, timeout=timeout)
+            response = self._session.get(url, stream=True, timeout=timeout)
         else:
-            response = session.get(url, stream=True)
+            response = self._session.get(url, stream=True)
 
         response.raise_for_status()
         return response.raw
diff --git a/pyvo/dal/scs.py b/pyvo/dal/scs.py
index 87ea955..e405757 100644
--- a/pyvo/dal/scs.py
+++ b/pyvo/dal/scs.py
@@ -87,7 +87,7 @@ class SCSService(DALService):
     a representation of a Cone Search service
     """
 
-    def __init__(self, baseurl):
+    def __init__(self, baseurl, session=None):
         """
         instantiate a Cone Search service
 
@@ -95,8 +95,10 @@ class SCSService(DALService):
         ----------
         baseurl : str
            the base URL for submitting search queries to the service.
+        session : object
+           optional session to use for network requests
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
 
     def _get_metadata(self):
         """
@@ -220,7 +222,7 @@ class SCSService(DALService):
         --------
         SCSQuery
         """
-        return SCSQuery(self.baseurl, pos, radius, verbosity, **keywords)
+        return SCSQuery(self.baseurl, pos, radius, verbosity, session=self._session, **keywords)
 
     def describe(self):
         print(self.description)
@@ -272,7 +274,7 @@ class SCSQuery(DALQuery):
     """
 
     def __init__(
-            self, baseurl, pos=None, radius=None, verbosity=None, **keywords):
+            self, baseurl, pos=None, radius=None, verbosity=None, session=None, **keywords):
         """
         initialize the query object with a baseurl and the given parameters
 
@@ -292,8 +294,10 @@ class SCSQuery(DALQuery):
             to return in the result table.  0 means the minimum
             set of columns, 3 means as many columns as are
             available.
+        session : object
+           optional session to use for network requests
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
 
         if pos is not None:
             self.pos = pos
@@ -406,7 +410,7 @@ class SCSQuery(DALQuery):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return SCSResults(self.execute_votable(), url=self.queryurl)
+        return SCSResults(self.execute_votable(), url=self.queryurl, session=self._session)
 
 
 class SCSResults(DALResults, DatalinkResultsMixin):
@@ -523,7 +527,7 @@ class SCSResults(DALResults, DatalinkResultsMixin):
         --------
         Record
         """
-        return SCSRecord(self, index)
+        return SCSRecord(self, index, session=self._session)
 
 
 class SCSRecord(DatalinkRecordMixin, Record):
diff --git a/pyvo/dal/sia.py b/pyvo/dal/sia.py
index c4c9a49..8f1e631 100644
--- a/pyvo/dal/sia.py
+++ b/pyvo/dal/sia.py
@@ -120,7 +120,7 @@ class SIAService(DALService):
     a representation of an SIA service
     """
 
-    def __init__(self, baseurl):
+    def __init__(self, baseurl, session=None):
         """
         instantiate an SIA service
 
@@ -128,8 +128,10 @@ class SIAService(DALService):
         ----------
         baseurl : str
            the base URL for submitting search queries to the service.
+        session : object
+           optional session to use for network requests
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
 
     def _get_metadata(self):
         """
@@ -313,7 +315,7 @@ class SIAService(DALService):
         SIAQuery
         """
         return SIAQuery(
-            self.baseurl, pos, size, format, intersect, verbosity, **keywords)
+            self.baseurl, pos, size, format, intersect, verbosity, self._session, **keywords)
 
     def describe(self):
         print(self.description)
@@ -342,7 +344,7 @@ class SIAQuery(DALQuery):
 
     def __init__(
             self, baseurl, pos=None, size=None, format=None, intersect=None,
-            verbosity=None, **keywords):
+            verbosity=None, session=None, **keywords):
         """
         initialize the query object with a baseurl and the given parameters
 
@@ -384,13 +386,15 @@ class SIAQuery(DALQuery):
             an integer value that indicates the volume of columns
             to return in the result table.  0 means the minimum
             set of columsn, 3 means as many columns as are  available.
+        session : object
+           optional session to use for network requests
         **keywords :
             additional parameters can be given via arbitrary
             case insensitive keyword arguments. Where there is overlap
             with the parameters set by the other arguments to
             this function, these keywords will override.
         """
-        super().__init__(baseurl, **keywords)
+        super().__init__(baseurl, session=session, **keywords)
 
         if pos:
             self.pos = pos
@@ -565,7 +569,7 @@ class SIAQuery(DALQuery):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return SIAResults(self.execute_votable(), url=self.queryurl)
+        return SIAResults(self.execute_votable(), url=self.queryurl, session=self._session)
 
 
 class SIAResults(DatalinkResultsMixin, DALResults):
@@ -643,7 +647,7 @@ class SIAResults(DatalinkResultsMixin, DALResults):
         --------
         Record
         """
-        return SIARecord(self, index)
+        return SIARecord(self, index, session=self._session)
 
 
 class SIARecord(SodaRecordMixin, DatalinkRecordMixin, Record):
diff --git a/pyvo/dal/sla.py b/pyvo/dal/sla.py
index 3f9fa6f..bec65c7 100644
--- a/pyvo/dal/sla.py
+++ b/pyvo/dal/sla.py
@@ -71,7 +71,7 @@ class SLAService(DALService):
     """
     a representation of an spectral line catalog (SLA) service
     """
-    def __init__(self, baseurl):
+    def __init__(self, baseurl, session=None):
         """
         instantiate an SLA service
 
@@ -79,8 +79,10 @@ class SLAService(DALService):
         ----------
         baseurl : str
            the base URL for submitting search queries to the service.
+        session : object
+           optional session to use for network requests
         """
-        super().__init__(baseurl,)
+        super().__init__(baseurl, session=session)
 
     def _get_metadata(self):
         """
@@ -187,7 +189,7 @@ class SLAService(DALService):
         --------
         SLAQuery
         """
-        return SLAQuery(self.baseurl, wavelength, request, **keywords)
+        return SLAQuery(self.baseurl, wavelength, request, session=self._session, **keywords)
 
     def describe(self):
         print(self.description)
@@ -239,7 +241,8 @@ class SLAQuery(DALQuery):
     """
 
     def __init__(
-            self, baseurl, wavelength=None, request="queryData", **keywords):
+            self, baseurl, wavelength=None, request="queryData", session=None,
+            **keywords):
         """
         initialize the query object with a baseurl and the given parameters
 
@@ -250,13 +253,15 @@ class SLAQuery(DALQuery):
         wavelength : `~astropy.units.Quantity` class or sequence of two floats
             the bandwidth range the observations belong to.
             assuming meters if unit is not specified.
+        session : object
+           optional session to use for network requests
         **keywords :
             additional parameters can be given via arbitrary
             case insensitive keyword arguments. Where there is overlap
             with the parameters set by the other arguments to
             this function, these keywords will override.
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
 
         if wavelength is not None:
             self.wavelength = wavelength
@@ -337,7 +342,7 @@ class SLAQuery(DALQuery):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return SLAResults(self.execute_votable(), self.queryurl)
+        return SLAResults(self.execute_votable(), self.queryurl, session=self._session)
 
 
 class SLAResults(DALResults):
@@ -416,7 +421,7 @@ class SLAResults(DALResults):
         --------
         Record
         """
-        return SLARecord(self, index)
+        return SLARecord(self, index, session=self._session)
 
 
 class SLARecord(Record):
diff --git a/pyvo/dal/ssa.py b/pyvo/dal/ssa.py
index 85de400..f1149c5 100644
--- a/pyvo/dal/ssa.py
+++ b/pyvo/dal/ssa.py
@@ -258,7 +258,7 @@ class SSAService(DALService):
         """
         return SSAQuery(
             self.baseurl, pos, diameter, band, time, format, request,
-            **keywords)
+            session=self._session, **keywords)
 
     def describe(self):
         print(self.description)
@@ -310,7 +310,7 @@ class SSAQuery(DALQuery):
 
     def __init__(
             self, baseurl, pos=None, diameter=None, band=None, time=None,
-            format=None, request="queryData", **keywords):
+            format=None, request="queryData", session=None, **keywords):
         """
         initialize the query object with a baseurl and the given parameters
 
@@ -336,13 +336,15 @@ class SSAQuery(DALQuery):
            graphical images (e.g. jpeg, png, gif; not FITS);
            "metadata" indicates that no images should be
            returned--only an empty table with complete metadata.
+        session : object
+           optional session to use for network requests
         **keywords :
            additional case insensitive parameters can be given via arbitrary
            case insensitive keyword arguments. Where there is overlap
            with the parameters set by the other arguments to
            this function, these keywords will override.
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
 
         if pos:
             self.pos = pos
@@ -565,7 +567,7 @@ class SSAQuery(DALQuery):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return SSAResults(self.execute_votable(), url=self.queryurl)
+        return SSAResults(self.execute_votable(), url=self.queryurl, session=self._session)
 
 
 class SSAResults(DatalinkResultsMixin, DALResults):
@@ -642,7 +644,7 @@ class SSAResults(DatalinkResultsMixin, DALResults):
         --------
         Record
         """
-        return SSARecord(self, index)
+        return SSARecord(self, index, session=self._session)
 
 
 class SSARecord(SodaRecordMixin, DatalinkRecordMixin, Record):
diff --git a/pyvo/dal/tap.py b/pyvo/dal/tap.py
index 8ddf70c..90433a8 100644
--- a/pyvo/dal/tap.py
+++ b/pyvo/dal/tap.py
@@ -21,7 +21,7 @@ from ..io import vosi, uws
 from ..io.vosi import tapregext as tr
 
 from ..utils.formatting import para_format_desc
-from ..utils.http import session
+from ..utils.http import use_session
 
 __all__ = [
     "search", "escape", "TAPService", "TAPQuery", "AsyncTAPJob", "TAPResults"]
@@ -77,7 +77,7 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
 
     _tables = None
 
-    def __init__(self, baseurl):
+    def __init__(self, baseurl, session=None):
         """
         instantiate a Tablee Access Protocol service
 
@@ -85,8 +85,17 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
         ----------
         baseurl :  str
            the base URL that should be used for forming queries to the service.
+        session : object
+           optional session to use for network requests
         """
-        super().__init__(baseurl)
+        super().__init__(baseurl, session=session)
+
+        # Check if the session has an update_from_capabilities attribute.
+        # This means that the session is aware of IVOA capabilities,
+        # and can use this information in processing network requests.
+        # One such usecase for this is auth.
+        if hasattr(self._session, 'update_from_capabilities'):
+            self._session.update_from_capabilities(self.capabilities)
 
     @property
     def tables(self):
@@ -96,7 +105,7 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
         if self._tables is None:
             tables_url = '{}/tables'.format(self.baseurl)
 
-            response = session.get(tables_url, stream=True)
+            response = self._session.get(tables_url, stream=True)
 
             try:
                 response.raise_for_status()
@@ -230,7 +239,7 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
         AsyncTAPJob
         """
         job = AsyncTAPJob.create(
-            self.baseurl, query, language, maxrec, uploads, **keywords)
+            self.baseurl, query, language, maxrec, uploads, self._session, **keywords)
         job = job.run().wait()
         job.raise_if_error()
         result = job.fetch_result()
@@ -267,7 +276,7 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
         AsyncTAPJob
         """
         return AsyncTAPJob.create(
-            self.baseurl, query, language, maxrec, uploads, **keywords)
+            self.baseurl, query, language, maxrec, uploads, self._session, **keywords)
 
     def create_query(
             self, query=None, mode="sync", language="ADQL", maxrec=None,
@@ -295,7 +304,7 @@ class TAPService(DALService, AvailabilityMixin, CapabilityMixin):
             a mapping from table names to objects containing a votable.
         """
         return TAPQuery(
-            self.baseurl, query, mode, language, maxrec, uploads, **keywords)
+            self.baseurl, query, mode, language, maxrec, uploads, self._session, **keywords)
 
     def describe(self, width=None):
         """
@@ -335,7 +344,7 @@ class AsyncTAPJob:
     @classmethod
     def create(
             cls, baseurl, query, language="ADQL", maxrec=None, uploads=None,
-            **keywords):
+            session=None, **keywords):
         """
         creates a async tap job on the server under `baseurl`
 
@@ -352,15 +361,17 @@ class AsyncTAPJob:
             the maximum records to return. defaults to the service default
         uploads : dict
             a mapping from table names to objects containing a votable
+        session : object
+           optional session to use for network requests
         """
         query = TAPQuery(
             baseurl, query, mode="async", language=language, maxrec=maxrec,
-            uploads=uploads, **keywords)
+            uploads=uploads, session=session, **keywords)
         response = query.submit()
-        job = cls(response.url)
+        job = cls(response.url, session=session)
         return job
 
-    def __init__(self, url):
+    def __init__(self, url, session=None):
         """
         initialize the job object with the given url and fetch remote values
 
@@ -370,6 +381,7 @@ class AsyncTAPJob:
             the job url
         """
         self._url = url
+        self._session = use_session(session)
         self._update()
 
     def __enter__(self):
@@ -393,13 +405,13 @@ class AsyncTAPJob:
         """
         try:
             if wait_for_statechange:
-                response = session.get(
+                response = self._session.get(
                     self.url, stream=True, timeout=timeout, params={
                         "WAIT": "-1"
                     }
                 )
             else:
-                response = session.get(self.url, stream=True, timeout=timeout)
+                response = self._session.get(self.url, stream=True, timeout=timeout)
             response.raise_for_status()
         except requests.RequestException as ex:
             raise DALServiceError.from_except(ex, self.url)
@@ -451,7 +463,7 @@ class AsyncTAPJob:
     @execution_duration.setter
     def execution_duration(self, value):
         try:
-            response = session.post(
+            response = self._session.post(
                 "{}/executionduration".format(self.url),
                 data={"EXECUTIONDURATION": str(value)})
             response.raise_for_status()
@@ -487,7 +499,7 @@ class AsyncTAPJob:
             pass
 
         try:
-            response = session.post(
+            response = self._session.post(
                 "{}/destruction".format(self.url),
                 data={"DESTRUCTION": value.strftime("%Y-%m-%dT%H:%M:%SZ")})
             response.raise_for_status()
@@ -526,7 +538,7 @@ class AsyncTAPJob:
     @query.setter
     def query(self, query):
         try:
-            response = session.post(
+            response = self._session.post(
                 '{}/parameters'.format(self.url),
                 data={"QUERY": query})
             response.raise_for_status()
@@ -547,7 +559,7 @@ class AsyncTAPJob:
         }
 
         try:
-            response = session.post(
+            response = self._session.post(
                 '{}/parameters'.format(self.url),
                 data={'UPLOAD': uploads.param()},
                 files=files
@@ -606,7 +618,7 @@ class AsyncTAPJob:
         starts the job / change phase to RUN
         """
         try:
-            response = session.post(
+            response = self._session.post(
                 '{}/phase'.format(self.url), data={"PHASE": "RUN"})
             response.raise_for_status()
         except requests.RequestException as ex:
@@ -619,7 +631,7 @@ class AsyncTAPJob:
         aborts the job / change phase to ABORT
         """
         try:
-            response = session.post(
+            response = self._session.post(
                 '{}/phase'.format(self.url), data={"PHASE": "ABORT"})
             response.raise_for_status()
         except requests.RequestException as ex:
@@ -674,7 +686,7 @@ class AsyncTAPJob:
         deletes the job. this object will become invalid.
         """
         try:
-            response = session.post(self.url, data={"ACTION": "DELETE"})
+            response = self._session.post(self.url, data={"ACTION": "DELETE"})
             response.raise_for_status()
         except requests.RequestException as ex:
             raise DALServiceError.from_except(ex, self.url)
@@ -698,7 +710,7 @@ class AsyncTAPJob:
         returns the result votable if query is finished
         """
         try:
-            response = session.get(self.result_uri, stream=True)
+            response = self._session.get(self.result_uri, stream=True)
             response.raise_for_status()
         except requests.RequestException as ex:
             self._update()
@@ -708,7 +720,7 @@ class AsyncTAPJob:
 
         response.raw.read = partial(
             response.raw.read, decode_content=True)
-        return TAPResults(votableparse(response.raw.read), url=self.result_uri)
+        return TAPResults(votableparse(response.raw.read), url=self.result_uri, session=self._session)
 
 
 class TAPQuery(DALQuery):
@@ -733,7 +745,7 @@ class TAPQuery(DALQuery):
 
     def __init__(
             self, baseurl, query, mode="sync", language="ADQL", maxrec=None,
-            uploads=None, **keywords):
+            uploads=None, session=None, **keywords):
         """
         initialize the query object with the given parameters
 
@@ -751,10 +763,12 @@ class TAPQuery(DALQuery):
             the amount of records to fetch
         uploads : dict
             Files to upload. Uses table name as key and table content as value.
+        session : object
+           optional session to use for network requests
         """
         baseurl = baseurl.rstrip("?")
 
-        super().__init__(baseurl, **keywords)
+        super().__init__(baseurl, session=session, **keywords)
 
         self._mode = mode if mode in ("sync", "async") else "sync"
         self._uploads = UploadList.fromdict(uploads or {})
@@ -807,7 +821,7 @@ class TAPQuery(DALQuery):
         DALFormatError
            for errors parsing the VOTable response
         """
-        return TAPResults(self.execute_votable(), url=self.queryurl)
+        return TAPResults(self.execute_votable(), url=self.queryurl, session=self._session)
 
     def submit(self):
         """
@@ -824,7 +838,7 @@ class TAPQuery(DALQuery):
             if upload.is_inline
         }
 
-        response = session.post(
+        response = self._session.post(
             url, data=self, stream=True, files=files)
         # requests doesn't decode the content by default
         response.raw.read = partial(response.raw.read, decode_content=True)
diff --git a/pyvo/dal/vosi.py b/pyvo/dal/vosi.py
index efc89ea..9ede6af 100644
--- a/pyvo/dal/vosi.py
+++ b/pyvo/dal/vosi.py
@@ -11,6 +11,7 @@ from .exceptions import DALServiceError
 from ..io import vosi
 from ..utils.url import url_sibling
 from ..utils.decorators import stream_decode_content, response_decode_content
+from ..utils.http import use_session
 
 __all__ = [
     'AvailabilityMixin', 'CapabilityMixin', 'VOSITables']
@@ -28,7 +29,7 @@ class AvailabilityMixin:
         """
         avail_url = '{}/availability'.format(self.baseurl)
 
-        response = requests.get(avail_url, stream=True)
+        response = self._session.get(avail_url, stream=True)
 
         try:
             response.raise_for_status()
@@ -73,7 +74,7 @@ class CapabilityMixin:
 
         for capa_url in capa_urls:
             try:
-                response = requests.get(capa_url, stream=True)
+                response = self._session.get(capa_url, stream=True)
                 response.raise_for_status()
                 break
             except requests.RequestException:
@@ -109,7 +110,7 @@ class TablesMixin(CapabilityMixin):
 
         for tables_url in tables_urls:
             try:
-                response = requests.get(tables_url, stream=True)
+                response = self._session.get(tables_url, stream=True)
                 response.raise_for_status()
                 break
             except requests.RequestException:
@@ -130,10 +131,11 @@ class VOSITables:
     Access to table names is like accessing dictionary keys. using iterator
     syntax or `keys()`
     """
-    def __init__(self, vosi_tables, endpoint_url):
+    def __init__(self, vosi_tables, endpoint_url, session=None):
         self._vosi_tables = vosi_tables
         self._endpoint_url = endpoint_url
         self._cache = {}
+        self._session = use_session(session)
 
     def __len__(self):
         return self._vosi_tables.ntables
@@ -167,7 +169,7 @@ class VOSITables:
 
     @response_decode_content
     def _get_table_file(self, tables_url):
-        return requests.get(tables_url, stream=True)
+        return self._session.get(tables_url, stream=True)
 
     def keys(self):
         """
diff --git a/pyvo/extensions/__init__.py b/pyvo/extensions/__init__.py
new file mode 100644
index 0000000..9dce85d
--- /dev/null
+++ b/pyvo/extensions/__init__.py
@@ -0,0 +1,1 @@
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
diff --git a/pyvo/extensions/auth/__init__.py b/pyvo/extensions/auth/__init__.py
new file mode 100644
index 0000000..9dce85d
--- /dev/null
+++ b/pyvo/extensions/auth/__init__.py
@@ -0,0 +1,1 @@
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
diff --git a/pyvo/extensions/auth/authsession.py b/pyvo/extensions/auth/authsession.py
new file mode 100644
index 0000000..841a29a
--- /dev/null
+++ b/pyvo/extensions/auth/authsession.py
@@ -0,0 +1,97 @@
+import logging
+
+from .authurls import AuthURLs
+from .credentialstore import CredentialStore
+
+
+class AuthSession:
+    """
+    A requests-like session that pyvo can use to dispatch its
+    network calls with authentication.
+
+    The user adds their credentials to the credentials object,
+    such as adding a cookie, certificate, or password.
+
+    The network requests made by pyvo pass through here, and
+    the URL of the request is matched against the capabilities
+    of the service.  Based on what credentials have been
+    provided and the capabilities of the service, appropriate
+    credentials are added to the request before it is sent.
+    """
+
+    def __init__(self):
+        super(AuthSession, self).__init__()
+        self.credentials = CredentialStore()
+        self._auth_urls = AuthURLs()
+
+    def add_security_method_for_url(self, url, security_method, exact=False):
+        """
+        Add a security method for a url.
+        This is additive with update_from_capabilities.  This
+        can be useful to set additional security methods that
+        aren't set in the capabilities for whatever reason.
+
+        Parameters
+        ----------
+        url : str
+            URL to set a security method for
+        security_method : str
+            URI of the security method to set
+        exact : bool
+            If True, match only this URL.  If false, match all URLs that
+            match this as a base URL.
+        """
+        self._auth_urls.add_security_method_for_url(url, security_method, exact=exact)
+
+    def update_from_capabilities(self, capabilities):
+        """
+        Update the URL to security method mapping using the
+        capabilities provided.
+
+        Parameters
+        ----------
+        capabilities : object
+            List of `~pyvo.io.vosi.voresource.Capabilities`
+        """
+        self._auth_urls.update_from_capabilities(capabilities)
+
+    def get(self, url, **kwargs):
+        """
+        Wrapper to make a HTTP GET request with authentication.
+        """
+        return self._request('GET', url, **kwargs)
+
+    def post(self, url, **kwargs):
+        """
+        Wrapper to make a HTTP POST request with authentication.
+        """
+        return self._request('POST', url, **kwargs)
+
+    def _request(self, http_method, url, **kwargs):
+        """
+        Make an HTTP request with authentication.
+
+        This function looks at the url of the request, determines
+        what credentials it should attach to the request to
+        authenticate, and then dispatches the request to the
+        underlying requests library using the session that
+        has been configured with the credentials.
+
+        Parameters
+        ----------
+        http_method : str
+            the HTTP verb of the request.
+        url : str
+            the URL to request
+        """
+        auth_methods = self._auth_urls.allowed_auth_methods(url)
+        logging.debug('Possible auth methods: %s', auth_methods)
+
+        negotiated_method = self.credentials.negotiate_method(auth_methods)
+        logging.debug('Using auth method: %s', negotiated_method)
+
+        session = self.credentials.get(negotiated_method)
+        return session.request(http_method, url, **kwargs)
+
+    def __repr__(self):
+        return '\n'.join([repr(self.credentials), repr(self._auth_urls)])
diff --git a/pyvo/extensions/auth/authurls.py b/pyvo/extensions/auth/authurls.py
new file mode 100644
index 0000000..42d77b0
--- /dev/null
+++ b/pyvo/extensions/auth/authurls.py
@@ -0,0 +1,114 @@
+import collections
+import logging
+
+import pyvo.extensions.auth.securitymethods as securitymethods
+
+
+class AuthURLs():
+    """
+    AuthURLs helps determine which security method should be used
+    with a given URL.  It learns the security methods through the
+    VOSI capabilities, which are passed in via update_from_capabilities.
+    """
+
+    def __init__(self):
+        self.full_urls = collections.defaultdict(set)
+        self.base_urls = collections.defaultdict(set)
+
+    def update_from_capabilities(self, capabilities):
+        """
+        Update the URL to security method mapping using the
+        capabilities provided.
+
+        Parameters
+        ----------
+        capabilities : object
+            List of `~pyvo.io.vosi.voresource.Capabilities`
+        """
+        for c in capabilities:
+            for i in c.interfaces:
+                for u in i.accessurls:
+                    url = u.content
+                    exact = u.use == 'full'
+
+                    if not i.securitymethods:
+                        self.add_security_method_for_url(url, securitymethods.ANONYMOUS, exact)
+
+                    for sm in i.securitymethods:
+                        method = sm.standardid or securitymethods.ANONYMOUS
+                        self.add_security_method_for_url(url, method, exact)
+
+    def add_security_method_for_url(self, url, security_method, exact=False):
+        """
+        Add a security method for a url.
+        This is additive with update_from_capabilities.  This
+        can be useful to set additional security methods that
+        aren't set in the capabilities for whatever reason.
+
+        Parameters
+        ----------
+        url : str
+            URL to set a security method for
+        security_method : str
+            URI of the security method to set
+        exact : bool
+            If True, match only this URL.  If false, match all URLs that
+            match this as a base URL.
+        """
+        if exact:
+            self.full_urls[url].add(security_method)
+        else:
+            self.base_urls[url].add(security_method)
+
+    def allowed_auth_methods(self, url):
+        """
+        Return the authentication methods allowed for a particular URL.
+        The methods are returned as URIs that represent security methods.
+
+        Parameters
+        ----------
+        url : str
+            the URL to determine authentication methods
+        """
+        logging.debug('Determining auth method for %s', url)
+
+        if url in self.full_urls:
+            methods = self.full_urls[url]
+            logging.debug('Matching full url %s, methods %s', url, methods)
+            return methods
+
+        for base_url, methods in self._iterate_base_urls():
+            if url.startswith(base_url):
+                logging.debug('Matching base url %s, methods %s', base_url, methods)
+                return methods
+
+        logging.debug('No match, using anonymous auth')
+        return {securitymethods.ANONYMOUS}
+
+    def _iterate_base_urls(self):
+        """
+        A generator to sort the base URLs in the correct way
+        to determine the most specific base_url.  This is done
+        by returning them longest to shortest.
+        """
+        def sort_by_len(x):
+            return len(x[0])
+
+        # Sort the base path matching URLs, so that
+        # the longest URLs (the most specific ones, if
+        # there is a tie) are used to determine the
+        # auth method.
+        for url, method in sorted(self.base_urls.items(),
+                                  key=sort_by_len,
+                                  reverse=True):
+            yield url, method
+
+    def __repr__(self):
+        urls = []
+
+        for url, methods in self.full_urls.items():
+            urls.append('Full match:' + url + ':' + str(methods))
+        for url, methods in self._iterate_base_urls():
+            urls.append('Base match:' + url + ':' + str(methods))
+
+        return '\n'.join(urls)
diff --git a/pyvo/extensions/auth/credentialstore.py b/pyvo/extensions/auth/credentialstore.py
new file mode 100644
index 0000000..76e3cb5
--- /dev/null
+++ b/pyvo/extensions/auth/credentialstore.py
@@ -0,0 +1,152 @@
+import logging
+
+from pyvo.utils.http import create_session
+
+import pyvo.extensions.auth.securitymethods as securitymethods
+
+
+class CredentialStore(object):
+    """
+    The credential store takes user credentials, and uses them
+    to create appropriate requests sessions for dispatching
+    requests using those credentials.
+
+    Different types of credentials can be passed in, such as
+    cookies, a jar of cookies, certificates, and basic auth.
+
+    A session can also be associated with a security method
+    URI by calling the set function.
+
+    Before a request is to be dispatched, the AuthSession
+    calls the get method to retrieve the appropriate
+    requests.Session for making that HTTP request.
+    """
+
+    def __init__(self):
+        self.credentials = {}
+        self.set(securitymethods.ANONYMOUS, create_session())
+
+    def negotiate_method(self, allowed_methods):
+        """
+        Compare the credentials provided by the user against the
+        security methods passed in, and determine which method is
+        to be used for making this request.
+
+        Parameters
+        ----------
+        allowed_methods : list(str)
+            list of allowed security methods to return
+
+        Raises
+        ------
+        Raises an exception if a common method could not be negotiated.
+        """
+        available_methods = set(self.credentials.keys())
+        methods = available_methods.intersection(allowed_methods)
+        logging.debug('Available methods: %s', methods)
+
+        # If we have no common auth mechanism, then fail.
+        if not methods:
+            msg = 'Negotiation failed.  Server supports %s, client supports %s' % \
+                (allowed_methods, available_methods)
+            raise Exception(msg)
+
+        # If there are more than 1 method to pick, don't pick
+        # anonymous over an actual method.
+        if len(methods) > 1:
+            methods.discard(securitymethods.ANONYMOUS)
+
+        # Pick a random method.
+        return methods.pop()
+
+    def set(self, method_uri, session):
+        """
+        Associate a security method URI with a requests.Session like object.
+
+        Parameters
+        ----------
+        method_uri : str
+            URI representing the security method
+        session : object
+            the requests.Session like object that will dispatch requests
+            for the authentication method provided by method_uri
+        """
+        self.credentials[method_uri] = session
+
+    def get(self, method_uri):
+        """
+        Retrieve the requests.Session like object associated with a security
+        method URI.
+
+        Parameters
+        ----------
+        method_uri : str
+            URI representing the security method
+        """
+        return self.credentials[method_uri]
+
+    def set_cookie(self, cookie_name, cookie_value, domain='', path='/'):
+        """
+        Add a cookie to use as authentication.
+
+        More than one call to set_cookie will add multiple cookies into
+        the same cookie jar used for the request.
+
+        Parameters
+        ----------
+        cookie_name : str
+            name of the cookie
+        cookie_value : str
+            value of the cookie
+        domain : str
+            restrict usage of this cookie to this domain
+        path : str
+            restrict usage of this cookie to this path
+        """
+        cookie_session = self.credentials.setdefault(securitymethods.COOKIE, create_session())
+        cookie_session.cookies.set(cookie_name, cookie_value, domain=domain, path=path)
+
+    def set_cookie_jar(self, cookie_jar):
+        """
+        Set the cookie jar to use for authentication.
+
+        Any previous cookies or cookie jars set will be removed.
+
+        Parameters
+        ----------
+        cookie_jar : obj
+            the cookie jar to use.
+        """
+        cookie_session = self.credentials.setdefault(securitymethods.COOKIE, create_session())
+        cookie_session.cookies = cookie_jar
+
+    def set_client_certificate(self, certificate_path):
+        """
+        Add a client certificate to use for authentication.
+
+        Parameters
+        ----------
+        certificate_path : str
+            path to the file of the client certificate
+        """
+        cert_session = create_session()
+        cert_session.cert = certificate_path
+        self.set(securitymethods.CLIENT_CERTIFICATE, cert_session)
+
+    def set_password(self, username, password):
+        """
+        Add a username / password for basic authentication.
+
+        Parameters
+        ----------
+        username : str
+            username to use
+        password : str
+            password to use
+        """
+        basic_session = create_session()
+        basic_session.auth = (username, password)
+        self.set(securitymethods.BASIC, basic_session)
+
+    def __repr__(self):
+        return 'Support for %s' % self.credentials.keys()
diff --git a/pyvo/extensions/auth/securitymethods.py b/pyvo/extensions/auth/securitymethods.py
new file mode 100644
index 0000000..48e3d16
--- /dev/null
+++ b/pyvo/extensions/auth/securitymethods.py
@@ -0,0 +1,4 @@
+ANONYMOUS = 'anonymous'
+BASIC = 'ivo://ivoa.net/sso#BasicAA'
+CLIENT_CERTIFICATE = 'ivo://ivoa.net/sso#tls-with-certificate'
+COOKIE = 'ivo://ivoa.net/sso#cookie'
diff --git a/pyvo/utils/http.py b/pyvo/utils/http.py
index 28995d6..796e179 100644
--- a/pyvo/utils/http.py
+++ b/pyvo/utils/http.py
@@ -5,15 +5,25 @@ HTTP utils
 import requests
 from ..version import version
 
+DEFAULT_USER_AGENT = 'python-pyvo/{}'.format(version)
 
-def requests_session(useragent=None):
-    if not useragent:
-        useragent = 'python-pyvo/{}'.format(version)
 
-    session = requests.Session()
-    session.headers['User-Agent'] = useragent
-
-    return session
+def use_session(session):
+    """
+    Return the session passed in, or create a default
+    session to use for this network request.
+    """
+    if session:
+        return session
+    else:
+        return create_session()
 
 
-session = requests_session()
+def create_session():
+    """
+    Create a new empty requests session with a pyvo
+    user agent.
+    """
+    session = requests.Session()
+    session.headers['User-Agent'] = DEFAULT_USER_AGENT
+    return session


Test patch:
diff --git a/pyvo/dal/tests/test_tap.py b/pyvo/dal/tests/test_tap.py
index def08b6..e3e22e5 100644
--- a/pyvo/dal/tests/test_tap.py
+++ b/pyvo/dal/tests/test_tap.py
@@ -3,12 +3,7 @@
 Tests for pyvo.dal.tap
 """
 from functools import partial
-
-try:
-    from contextlib import ExitStack
-except ImportError:
-    from contextlib2 import ExitStack
-
+from contextlib import ExitStack
 import re
 from io import BytesIO
 from urllib.parse import parse_qsl
@@ -54,144 +49,152 @@ def sync_fixture(mocker):
         yield matcher
 
 
-@pytest.fixture()
-def async_fixture(mocker):
-    class Callback:
-        def __init__(self):
-            self._jobs = dict()
-
-        def create(self, request, context):
-            newid = max(list(self._jobs.keys()) or [0]) + 1
+class MockAsyncTAPServer:
+    def __init__(self):
+        self._jobs = dict()
+
+    def validator(self, request):
+        pass
+
+    def use(self, mocker):
+        with ExitStack() as stack:
+            matchers = {
+                'create': stack.enter_context(mocker.register_uri(
+                    'POST', 'http://example.com/tap/async',
+                    content=self.create
+                )),
+                'job': stack.enter_context(mocker.register_uri(
+                    requests_mock.ANY, job_re_path_full, content=self.job
+                )),
+                'phase': stack.enter_context(mocker.register_uri(
+                    requests_mock.ANY, job_re_phase_full, content=self.phase
+                )),
+                'parameters': stack.enter_context(mocker.register_uri(
+                    requests_mock.ANY, job_re_parameters_full,
+                    content=self.parameters
+                )),
+                'result': stack.enter_context(mocker.register_uri(
+                    'GET', job_re_result_full, content=self.result
+                ))
+            }
+            yield matchers
+
+    def create(self, request, context):
+        self.validator(request)
+        newid = max(list(self._jobs.keys()) or [0]) + 1
+        data = dict(parse_qsl(request.body))
+
+        job = JobFile()
+        job.version = "1.1"
+        job.jobid = newid
+        job.phase = 'PENDING'
+        job.quote = Time.now() + TimeDelta(1, format='sec')
+        job.creationtime = Time.now()
+        job.executionduration = TimeDelta(3600, format='sec')
+        job.destruction = Time.now() + TimeDelta(3600, format='sec')
+
+        for key, value in data.items():
+            param = Parameter(id=key.lower())
+            param.content = value
+            job.parameters.append(param)
+
+        context.status_code = 303
+        context.reason = 'See other'
+        context.headers['Location'] = (
+            'http://example.com/tap/async/{}'.format(newid))
+
+        self._jobs[newid] = job
+
+    def job(self, request, context):
+        self.validator(request)
+        jobid = int(job_re_path.match(request.path).group(1))
+
+        if request.method == 'GET':
+            job = self._jobs[jobid]
+            io = BytesIO()
+            job.to_xml(io)
+            return io.getvalue()
+        elif request.method == 'POST':
             data = dict(parse_qsl(request.body))
+            action = data.get('ACTION')
 
-            job = JobFile()
-            job.version = "1.1"
-            job.jobid = newid
-            job.phase = 'PENDING'
-            job.quote = Time.now() + TimeDelta(1, format='sec')
-            job.creationtime = Time.now()
-            job.executionduration = TimeDelta(3600, format='sec')
-            job.destruction = Time.now() + TimeDelta(3600, format='sec')
-
-            for key, value in data.items():
-                param = Parameter(id=key.lower())
-                param.content = value
-                job.parameters.append(param)
-
-            context.status_code = 303
-            context.reason = 'See other'
-            context.headers['Location'] = (
-                'http://example.com/tap/async/{}'.format(newid))
-
-            self._jobs[newid] = job
-
-        def job(self, request, context):
-            jobid = int(job_re_path.match(request.path).group(1))
-
-            if request.method == 'GET':
-                job = self._jobs[jobid]
-                io = BytesIO()
-                job.to_xml(io)
-                return io.getvalue()
-            elif request.method == 'POST':
-                data = dict(parse_qsl(request.body))
-                action = data.get('ACTION')
-
-                if action == 'DELETE':
-                    del self._jobs[jobid]
-
-        def phase(self, request, context):
-            jobid = int(job_re_path.match(request.path).group(1))
-
-            if request.method == 'GET':
-                phase = self._jobs[jobid].phase
-                return phase
-            elif request.method == 'POST':
-                newphase = request.body.split('=')[-1]
-                job = self._jobs[jobid]
-                result = get_pkg_data_contents('data/tap/obscore-image.xml')
-
-                if newphase == 'RUN':
-                    newphase = 'COMPLETED'
-                    result = Result(**{
-                        'id': 'result',
-                        'size': len(result),
-                        'mime-type': 'application/x-votable+xml',
-                        'xlink:href': (
-                            'http://example.com/tap/async/{}/results/result'
-                        ).format(jobid)
-                    })
-
-                    try:
-                        job.results[0] = result
-                    except (IndexError, TypeError):
-                        job.results.append(result)
-
-                job.phase = newphase
-
-        def parameters(self, request, context):
-            jobid = int(job_re_path.match(request.path).group(1))
-            job = self._jobs[jobid]
+            if action == 'DELETE':
+                del self._jobs[jobid]
 
-            if request.method == 'GET':
-                pass
-            elif request.method == 'POST':
-                data = dict(parse_qsl(request.body))
-
-                if 'QUERY' in data:
-                    assert data['QUERY'] == 'SELECT TOP 42 * FROM ivoa.obsCore'
-                    for param in job.parameters:
-                        if param.id_ == 'query':
-                            param.content = data['QUERY']
-                if 'UPLOAD' in data:
-                    for param in job.parameters:
-                        if param.id_ == 'upload':
-                            uploads1 = {data[0]: data[1] for data in [
-                                data.split(',') for data
-                                in data['UPLOAD'].split(';')
-                            ]}
-
-                            uploads2 = {data[0]: data[1] for data in [
-                                data.split(',') for data
-                                in param.content.split(';')
-                            ]}
-
-                            uploads1.update(uploads2)
-
-                            param.content = ';'.join([
-                                '{}={}'.format(key, value) for key, value
-                                in uploads1.items()
-                            ])
-
-        def result(self, request, context):
-            # jobid = int(job_re_path.match(request.path)[1])
-            return get_pkg_data_contents('data/tap/obscore-image.xml')
-
-    callback = Callback()
+    def phase(self, request, context):
+        self.validator(request)
+        jobid = int(job_re_path.match(request.path).group(1))
 
-    with ExitStack() as stack:
-        matchers = {
-            'create': stack.enter_context(mocker.register_uri(
-                'POST', 'http://example.com/tap/async',
-                content=callback.create
-            )),
-            'job': stack.enter_context(mocker.register_uri(
-                requests_mock.ANY, job_re_path_full, content=callback.job
-            )),
-            'phase': stack.enter_context(mocker.register_uri(
-                requests_mock.ANY, job_re_phase_full, content=callback.phase
-            )),
-            'parameters': stack.enter_context(mocker.register_uri(
-                requests_mock.ANY, job_re_parameters_full,
-                content=callback.parameters
-            )),
+        if request.method == 'GET':
+            phase = self._jobs[jobid].phase
+            return phase
+        elif request.method == 'POST':
+            newphase = request.body.split('=')[-1]
+            job = self._jobs[jobid]
+            result = get_pkg_data_contents('data/tap/obscore-image.xml')
+
+            if newphase == 'RUN':
+                newphase = 'COMPLETED'
+                result = Result(**{
+                    'id': 'result',
+                    'size': len(result),
+                    'mime-type': 'application/x-votable+xml',
+                    'xlink:href': (
+                        'http://example.com/tap/async/{}/results/result'
+                    ).format(jobid)
+                })
+
+                try:
+                    job.results[0] = result
+                except (IndexError, TypeError):
+                    job.results.append(result)
+
+            job.phase = newphase
+
+    def parameters(self, request, context):
+        self.validator(request)
+        jobid = int(job_re_path.match(request.path).group(1))
+        job = self._jobs[jobid]
+
+        if request.method == 'GET':
+            pass
+        elif request.method == 'POST':
+            data = dict(parse_qsl(request.body))
 
-            'result': stack.enter_context(mocker.register_uri(
-                'GET', job_re_result_full, content=callback.result
-            ))
-        }
+            if 'QUERY' in data:
+                assert data['QUERY'] == 'SELECT TOP 42 * FROM ivoa.obsCore'
+                for param in job.parameters:
+                    if param.id_ == 'query':
+                        param.content = data['QUERY']
+            if 'UPLOAD' in data:
+                for param in job.parameters:
+                    if param.id_ == 'upload':
+                        uploads1 = {data[0]: data[1] for data in [
+                            data.split(',') for data
+                            in data['UPLOAD'].split(';')
+                        ]}
+
+                        uploads2 = {data[0]: data[1] for data in [
+                            data.split(',') for data
+                            in param.content.split(';')
+                        ]}
+
+                        uploads1.update(uploads2)
+
+                        param.content = ';'.join([
+                            '{}={}'.format(key, value) for key, value
+                            in uploads1.items()
+                        ])
+
+    def result(self, request, context):
+        self.validator(request)
+        return get_pkg_data_contents('data/tap/obscore-image.xml')
 
-        yield matchers
+
+@pytest.fixture()
+def async_fixture(mocker):
+    mock_server = MockAsyncTAPServer()
+    yield from mock_server.use(mocker)
 
 
 @pytest.fixture()
diff --git a/pyvo/extensions/auth/tests/__init__.py b/pyvo/extensions/auth/tests/__init__.py
new file mode 100644
index 0000000..9dce85d
--- /dev/null
+++ b/pyvo/extensions/auth/tests/__init__.py
@@ -0,0 +1,1 @@
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
diff --git a/pyvo/extensions/auth/tests/conftest.py b/pyvo/extensions/auth/tests/conftest.py
new file mode 100644
index 0000000..f6612eb
--- /dev/null
+++ b/pyvo/extensions/auth/tests/conftest.py
@@ -0,0 +1,29 @@
+from contextlib import contextmanager
+
+import pytest
+import requests_mock
+
+
+class ContextAdapter(requests_mock.Adapter):
+    """
+    requests_mock adapter where ``register_uri`` returns a context manager
+    """
+    @contextmanager
+    def register_uri(self, *args, **kwargs):
+        matcher = super().register_uri(*args, **kwargs)
+
+        yield matcher
+
+        self.remove_matcher(matcher)
+
+    def remove_matcher(self, matcher):
+        if matcher in self._matchers:
+            self._matchers.remove(matcher)
+
+
+@pytest.fixture(scope='session')
+def mocker():
+    with requests_mock.Mocker(
+        adapter=ContextAdapter(case_sensitive=True)
+    ) as mocker:
+        yield mocker
diff --git a/pyvo/extensions/auth/tests/data/tap/capabilities.xml b/pyvo/extensions/auth/tests/data/tap/capabilities.xml
new file mode 100644
index 0000000..0a4dbc5
--- /dev/null
+++ b/pyvo/extensions/auth/tests/data/tap/capabilities.xml
@@ -0,0 +1,102 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<vosi:capabilities xmlns:vosi="http://www.ivoa.net/xml/VOSICapabilities/v1.0" xmlns:vs="http://www.ivoa.net/xml/VODataService/v1.1" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <capability standardID="ivo://ivoa.net/std/VOSI#capabilities">
+    <interface xsi:type="vs:ParamHTTP" role="std">
+      <accessURL use="full">http://example.com/tap/capabilities</accessURL>
+    </interface>
+  </capability>
+  <capability standardID="ivo://ivoa.net/std/VOSI#availability">
+    <interface xsi:type="vs:ParamHTTP" role="std">
+      <accessURL use="full">http://example.com/tap/availability</accessURL>
+    </interface>
+  </capability>
+  <capability standardID="ivo://ivoa.net/std/VOSI#tables-1.1">
+    <interface xsi:type="vs:ParamHTTP" role="std" version="1.1">
+      <accessURL use="base">http://example.com/tap/tables</accessURL>
+      $security_methods
+    </interface>
+  </capability>
+  <!-- TAP-1.1 -->
+  <capability xmlns:tr="http://www.ivoa.net/xml/TAPRegExt/v1.0" standardID="ivo://ivoa.net/std/TAP" xsi:type="tr:TableAccess">
+    <interface xsi:type="vs:ParamHTTP" role="std" version="1.1">
+      <accessURL use="base">http://example.com/tap</accessURL>
+      $security_methods
+    </interface>
+    <language>
+      <name>ADQL</name>
+      <version ivo-id="ivo://ivoa.net/std/ADQL#v2.0">2.0</version>
+      <description>ADQL-2.0</description>
+      <languageFeatures type="ivo://ivoa.net/std/TAPRegExt#features-adqlgeo">
+        <feature>
+          <form>POINT</form>
+        </feature>
+        <feature>
+          <form>CIRCLE</form>
+        </feature>
+        <feature>
+          <form>DISTANCE</form>
+        </feature>
+        <feature>
+          <form>POLYGON</form>
+        </feature>
+        <feature>
+          <form>REGION</form>
+        </feature>
+        <feature>
+          <form>CONTAINS</form>
+        </feature>
+        <feature>
+          <form>INTERSECTS</form>
+        </feature>
+        <feature>
+          <form>AREA</form>
+        </feature>
+        <feature>
+          <form>CENTROID</form>
+        </feature>
+        <feature>
+          <form>COORD1</form>
+        </feature>
+        <feature>
+          <form>COORD2</form>
+        </feature>
+      </languageFeatures>
+    </language>
+    <outputFormat ivo-id="ivo://ivoa.net/std/TAPRegExt#output-votable-td">
+      <mime>application/x-votable+xml</mime>
+      <alias>votable</alias>
+    </outputFormat>
+    <outputFormat ivo-id="ivo://ivoa.net/std/TAPRegExt#output-votable-td">
+      <mime>text/xml</mime>
+    </outputFormat>
+    <outputFormat>
+      <mime>text/csv</mime>
+      <alias>csv</alias>
+    </outputFormat>
+    <outputFormat>
+      <mime>text/tab-separated-values</mime>
+      <alias>tsv</alias>
+    </outputFormat>
+    <uploadMethod ivo-id="ivo://ivoa.net/std/TAPRegExt#upload-inline" />
+    <uploadMethod ivo-id="ivo://ivoa.net/std/TAPRegExt#upload-http" />
+    <uploadMethod ivo-id="ivo://ivoa.net/std/TAPRegExt#upload-https" />
+    <retentionPeriod>
+      <default>604800</default>
+      <hard>604800</hard>
+    </retentionPeriod>
+    <executionDuration>
+      <default>600</default>
+      <hard>600</hard>
+    </executionDuration>
+    <!-- outputLimit for async queries: 128MB -->
+    <outputLimit>
+      <default unit="byte">134217728</default>
+      <hard unit="byte">134217728</hard>
+    </outputLimit>
+    <!-- outputLimit for sync queries: no limit -->
+    <uploadLimit>
+      <default unit="row">10000</default>
+      <hard unit="row">10000</hard>
+    </uploadLimit>
+  </capability>
+</vosi:capabilities>
diff --git a/pyvo/extensions/auth/tests/setup_package.py b/pyvo/extensions/auth/tests/setup_package.py
new file mode 100644
index 0000000..d4c2d27
--- /dev/null
+++ b/pyvo/extensions/auth/tests/setup_package.py
@@ -0,0 +1,7 @@
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+import os
+
+
+def get_package_data():
+    paths = [os.path.join('data/tap', '*.xml')]
+    return {'pyvo.extensions.auth.tests': paths}
diff --git a/pyvo/extensions/auth/tests/test_auth.py b/pyvo/extensions/auth/tests/test_auth.py
new file mode 100644
index 0000000..eb1c12e
--- /dev/null
+++ b/pyvo/extensions/auth/tests/test_auth.py
@@ -0,0 +1,174 @@
+#!/usr/bin/env python
+# Licensed under a 3-clause BSD style license - see LICENSE.rst
+"""
+Tests for pyvo.extensions.auth
+"""
+import base64
+from requests.cookies import RequestsCookieJar
+from string import Template
+
+import pytest
+
+from astropy.utils.data import get_pkg_data_contents
+
+import pyvo.dal
+from pyvo.extensions.auth.authsession import AuthSession
+
+from pyvo.dal.tests.test_tap import MockAsyncTAPServer
+
+
+@pytest.fixture()
+def security_methods():
+    return [None]
+
+
+@pytest.fixture()
+def auth_capabilities(mocker, security_methods):
+    anon_element = '<securityMethod />'
+    sm_element = Template('<securityMethod standardID="$m" />')
+    method_elements = []
+
+    for m in security_methods:
+        if m is None:
+            method_elements.append(anon_element)
+        else:
+            method_elements.append(sm_element.substitute(m=m))
+
+    def callback(request, context):
+        t = Template(get_pkg_data_contents('data/tap/capabilities.xml'))
+        capabilities = t.substitute(security_methods=method_elements)
+        return capabilities.encode('utf-8')
+
+    with mocker.register_uri(
+        'GET', 'http://example.com/tap/capabilities', content=callback
+    ) as matcher:
+        yield matcher
+
+
+class MockAnonAuthTAPServer(MockAsyncTAPServer):
+    def validator(self, request):
+        assert request.cert is None
+        assert 'Authorization' not in request.headers
+        assert 'Cookie' not in request.headers
+
+
+@pytest.fixture()
+def anon_auth_service(mocker):
+    yield from MockAnonAuthTAPServer().use(mocker)
+
+
+class MockCookieAuthTAPServer(MockAsyncTAPServer):
+    def validator(self, request):
+        assert request.cert is None
+        assert 'Authorization' not in request.headers
+        assert request.headers.get('Cookie', None) == 'TEST_COOKIE=BADCOOKIE'
+
+
+@pytest.fixture()
+def cookie_auth_service(mocker):
+    yield from MockCookieAuthTAPServer().use(mocker)
+
+
+class MockCertificateAuthTAPServer(MockAsyncTAPServer):
+    def validator(self, request):
+        assert request.cert == 'client-certificate.pem'
+        assert 'Authorization' not in request.headers
+        assert 'Cookie' not in request.headers
+
+
+@pytest.fixture()
+def certificate_auth_service(mocker):
+    yield from MockCertificateAuthTAPServer().use(mocker)
+
+
+class MockBasicAuthTAPServer(MockAsyncTAPServer):
+    def validator(self, request):
+        pw = 'testuser:hunter2'.encode('ascii')
+        basic_encoded = 'Basic ' + base64.b64encode(pw).decode('ascii')
+
+        assert request.cert is None
+        assert request.headers.get('Authorization', None) == basic_encoded
+        assert 'Cookie' not in request.headers
+
+
+@pytest.fixture()
+def basic_auth_service(mocker):
+    yield from MockBasicAuthTAPServer().use(mocker)
+
+
+@pytest.mark.usefixtures('cookie_auth_service', 'auth_capabilities')
+@pytest.mark.parametrize('security_methods', [[None, 'ivo://ivoa.net/sso#cookie']])
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_cookies_auth():
+    session = AuthSession()
+    session.credentials.set_cookie('TEST_COOKIE', 'BADCOOKIE')
+    service = pyvo.dal.TAPService('http://example.com/tap', session)
+    service.run_async("SELECT * FROM ivoa.obscore")
+
+
+@pytest.mark.usefixtures('cookie_auth_service', 'auth_capabilities')
+@pytest.mark.parametrize('security_methods', [[None, 'ivo://ivoa.net/sso#cookie']])
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_cookie_jar_auth():
+    session = AuthSession()
+    jar = RequestsCookieJar()
+    jar.set('TEST_COOKIE', 'BADCOOKIE')
+    session.credentials.set_cookie_jar(jar)
+    service = pyvo.dal.TAPService('http://example.com/tap', session)
+    service.run_async("SELECT * FROM ivoa.obscore")
+
+
+@pytest.mark.usefixtures('certificate_auth_service', 'auth_capabilities')
+@pytest.mark.parametrize('security_methods', [[None, 'ivo://ivoa.net/sso#tls-with-certificate']])
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_certificate_auth():
+    session = AuthSession()
+    session.credentials.set_client_certificate('client-certificate.pem')
+    service = pyvo.dal.TAPService('http://example.com/tap', session)
+    service.run_async("SELECT * FROM ivoa.obscore")
+
+
+@pytest.mark.usefixtures('basic_auth_service', 'auth_capabilities')
+@pytest.mark.parametrize('security_methods', [[None, 'ivo://ivoa.net/sso#BasicAA']])
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_basic_auth():
+    session = AuthSession()
+    session.credentials.set_password('testuser', 'hunter2')
+    service = pyvo.dal.TAPService('http://example.com/tap', session)
+    service.run_async("SELECT * FROM ivoa.obscore")
+
+
+@pytest.mark.usefixtures('basic_auth_service', 'auth_capabilities')
+@pytest.mark.parametrize('security_methods', [[None, 'ivo://ivoa.net/sso#BasicAA']])
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_negotiation():
+    session = AuthSession()
+    session.credentials.set_password('testuser', 'hunter2')
+    session.credentials.set_client_certificate('client-certificate.pem')
+    session.credentials.set_cookie('TEST_COOKIE', 'BADCOOKIE')
+    service = pyvo.dal.TAPService('http://example.com/tap', session)
+    service.run_async("SELECT * FROM ivoa.obscore")
+
+
+@pytest.mark.usefixtures('anon_auth_service', 'auth_capabilities')
+@pytest.mark.parametrize('security_methods', [[None, 'ivo://ivoa.net/sso#FancyAuth']])
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W27")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W48")
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_no_common_auth_negotiation():
+    session = AuthSession()
+    session.credentials.set_password('testuser', 'hunter2')
+    session.credentials.set_client_certificate('client-certificate.pem')
+    session.credentials.set_cookie('TEST_COOKIE', 'BADCOOKIE')
+    service = pyvo.dal.TAPService('http://example.com/tap', session)
+    service.run_async("SELECT * FROM ivoa.obscore")