sjs/vdirsyncer
1
0
Fork 0
mirror of https://github.com/samsonjs/vdirsyncer.git synced 2026-03-25 08:55:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

A lot of module restructuring

Browse source 
- Split utils up again
- Optimize performance when importing a specific storage. This is useful
  for khal which uses our FilesystemStorage (and doesn't want to import
  requests).
This commit is contained in:
Markus Unterwaditzer 2015-04-13 17:33:44 +02:00
parent 2320570217
commit e2eb79d656
9 changed files with 285 additions and 270 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
tests/cli/test_config.py
View file

@ -61,7 +61,7 @@ def test_storage_instance_from_config(monkeypatch):
return 'OK'
import vdirsyncer.storage
monkeypatch.setitem(vdirsyncer.storage.storage_names, 'lol', lol)
monkeypatch.setitem(vdirsyncer.cli.utils.storage_names, 'lol', lol)
config = {'type': 'lol', 'foo': 'bar', 'baz': 1}
assert cli.utils.storage_instance_from_config(config) == 'OK'

40
tests/utils/test_main.py
View file

@ -12,6 +12,8 @@ import pytest
import requests
import vdirsyncer.doubleclick as doubleclick
import vdirsyncer.utils.http
import vdirsyncer.utils.password
import vdirsyncer.utils as utils
from .. import blow_up
@ -33,7 +35,7 @@ class EmptyKeyring(object):
@pytest.fixture(autouse=True)
def empty_password_storages(monkeypatch):
monkeypatch.setattr('netrc.netrc', EmptyNetrc)
monkeypatch.setattr(utils, 'keyring', EmptyKeyring())
monkeypatch.setattr(utils.password, 'keyring', EmptyKeyring())
def test_get_password_from_netrc(monkeypatch):
@ -52,7 +54,7 @@ def test_get_password_from_netrc(monkeypatch):
monkeypatch.setattr('netrc.netrc', Netrc)
monkeypatch.setattr('getpass.getpass', blow_up)
_password = utils.get_password(username, resource)
_password = utils.password.get_password(username, resource)
assert _password == password
assert calls == [hostname]
@ -66,14 +68,14 @@ def test_get_password_from_system_keyring(monkeypatch):
class KeyringMock(object):
def get_password(self, resource, _username):
assert _username == username
assert resource == utils.password_key_prefix + hostname
assert resource == utils.password.password_key_prefix + hostname
return password
monkeypatch.setattr(utils, 'keyring', KeyringMock())
monkeypatch.setattr(utils.password, 'keyring', KeyringMock())
monkeypatch.setattr('getpass.getpass', blow_up)
_password = utils.get_password(username, resource)
_password = utils.password.get_password(username, resource)
assert _password == password
@ -98,7 +100,7 @@ def test_get_password_from_command(tmpdir):
@doubleclick.click.pass_context
def fake_app(ctx):
ctx.obj = {'config': ({'password_command': filepath}, {}, {})}
_password = utils.get_password(username, resource)
_password = utils.password.get_password(username, resource)
assert _password == password
runner = CliRunner()
@ -112,7 +114,7 @@ def test_get_password_from_prompt():
@click.command()
def fake_app():
x = utils.get_password(user, resource)
x = utils.password.get_password(user, resource)
click.echo('Password is {}'.format(x))
runner = CliRunner()
@ -127,22 +129,24 @@ def test_get_password_from_prompt():
def test_set_keyring_password(monkeypatch):
class KeyringMock(object):
def get_password(self, resource, username):
assert resource == utils.password_key_prefix + 'example.com'
assert resource == \
utils.password.password_key_prefix + 'example.com'
assert username == 'foouser'
return None
def set_password(self, resource, username, password):
assert resource == utils.password_key_prefix + 'example.com'
assert resource == \
utils.password.password_key_prefix + 'example.com'
assert username == 'foouser'
assert password == 'hunter2'
monkeypatch.setattr(utils, 'keyring', KeyringMock())
monkeypatch.setattr(utils.password, 'keyring', KeyringMock())
@doubleclick.click.command()
@doubleclick.click.pass_context
def fake_app(ctx):
ctx.obj = {}
x = utils.get_password('foouser', 'http://example.com/a/b')
x = utils.password.get_password('foouser', 'http://example.com/a/b')
click.echo('password is ' + x)
runner = CliRunner()
@ -163,12 +167,12 @@ def test_get_password_from_cache(monkeypatch):
@doubleclick.click.pass_context
def fake_app(ctx):
ctx.obj = {}
x = utils.get_password(user, resource)
x = utils.password.get_password(user, resource)
click.echo('Password is {}'.format(x))
monkeypatch.setattr(doubleclick.click, 'prompt', blow_up)
assert (user, 'example.com') in ctx.obj['passwords']
x = utils.get_password(user, resource)
x = utils.password.get_password(user, resource)
click.echo('Password is {}'.format(x))
runner = CliRunner()
@ -212,13 +216,13 @@ def test_request_ssl(httpsserver):
httpsserver.serve_content('') # we need to serve something
with pytest.raises(requests.exceptions.SSLError) as excinfo:
utils.request('GET', httpsserver.url)
utils.http.request('GET', httpsserver.url)
assert 'certificate verify failed' in str(excinfo.value)
utils.request('GET', httpsserver.url, verify=False)
utils.request('GET', httpsserver.url,
utils.http.request('GET', httpsserver.url, verify=False)
utils.http.request('GET', httpsserver.url,
verify_fingerprint=sha1)
utils.request('GET', httpsserver.url, verify_fingerprint=md5)
utils.http.request('GET', httpsserver.url, verify_fingerprint=md5)
with pytest.raises(requests.exceptions.SSLError) as excinfo:
utils.request('GET', httpsserver.url,
utils.http.request('GET', httpsserver.url,
verify_fingerprint=''.join(reversed(sha1)))
assert 'Fingerprints did not match' in str(excinfo.value)

28
vdirsyncer/cli/utils.py
View file

@ -14,7 +14,6 @@ from atomicwrites import atomic_write
from .. import DOCS_HOME, PROJECT_HOME, exceptions, log
from ..doubleclick import click
from ..storage import storage_names
from ..sync import IdentConflict, StorageEmpty, SyncConflict
from ..utils import expand_path, get_class_init_args
from ..utils.compat import text_type
@ -32,6 +31,33 @@ except ImportError:
import queue
def _generate_storage_dict():
from ..storage.dav import CaldavStorage, CarddavStorage
from ..storage.filesystem import FilesystemStorage
from ..storage.http import HttpStorage
from ..storage.singlefile import SingleFileStorage
classes = (
CaldavStorage,
CarddavStorage,
FilesystemStorage,
HttpStorage,
SingleFileStorage
)
rv = {}
for cls in classes:
key = cls.storage_name
assert key
assert isinstance(key, str)
assert key not in rv
rv[key] = cls
return rv
storage_names = _generate_storage_dict()
del _generate_storage_dict
cli_logger = log.get(__name__)
GENERAL_ALL = frozenset(['status_path', 'password_command'])

26
vdirsyncer/storage/__init__.py
View file

@ -5,29 +5,3 @@ offer basic CRUD-ish methods for modifying those collections. The exact
interface is described in `vdirsyncer.storage.base`, the `Storage` class should
be a superclass of all storage classes.
'''
from .dav import CaldavStorage, CarddavStorage
from .filesystem import FilesystemStorage
from .http import HttpStorage
from .singlefile import SingleFileStorage
def _generate_storage_dict(*classes):
rv = {}
for cls in classes:
key = cls.storage_name
assert key
assert isinstance(key, str)
assert key not in rv
rv[key] = cls
return rv
storage_names = _generate_storage_dict(
CaldavStorage,
CarddavStorage,
FilesystemStorage,
HttpStorage,
SingleFileStorage
)
del _generate_storage_dict

20
vdirsyncer/storage/dav.py
View file

@ -4,7 +4,7 @@ import datetime
from lxml import etree
from requests import session as requests_session
import requests
from requests.exceptions import HTTPError
from .base import Item, Storage
@ -24,8 +24,8 @@ def _normalize_href(base, href):
schema.'''
if not href:
raise ValueError(href)
x = utils.urlparse.urljoin(base, href)
x = utils.urlparse.urlsplit(x).path
x = utils.compat.urlparse.urljoin(base, href)
x = utils.compat.urlparse.urlsplit(x).path
return x
@ -121,7 +121,7 @@ class Discover(object):
rv = root.find('.//{DAV:}current-user-principal/{DAV:}href')
if rv is None:
raise InvalidXMLResponse()
return utils.urlparse.urljoin(response.url, rv.text)
return utils.compat.urlparse.urljoin(response.url, rv.text)
def find_home(self, url=None):
if url is None:
@ -137,7 +137,7 @@ class Discover(object):
rv = root.find('.//' + self._homeset_tag + '/{*}href')
if rv is None:
raise InvalidXMLResponse()
return utils.urlparse.urljoin(response.url, rv.text)
return utils.compat.urlparse.urljoin(response.url, rv.text)
def find_collections(self, url=None):
if url is None:
@ -157,7 +157,7 @@ class Discover(object):
href = response.find('{*}href')
if href is None:
raise InvalidXMLResponse()
href = utils.urlparse.urljoin(r.url, href.text)
href = utils.compat.urlparse.urljoin(r.url, href.text)
if href not in done:
done.add(href)
yield {'href': href, 'displayname': displayname}
@ -221,19 +221,19 @@ class DavSession(object):
}
self.useragent = useragent
self.url = url.rstrip('/') + '/'
self.parsed_url = utils.urlparse.urlparse(self.url)
self.parsed_url = utils.compat.urlparse.urlparse(self.url)
self._session = None
def request(self, method, path, **kwargs):
url = self.url
if path:
url = utils.urlparse.urljoin(self.url, path)
url = utils.compat.urlparse.urljoin(self.url, path)
if self._session is None:
self._session = requests_session()
self._session = requests.session()
more = dict(self._settings)
more.update(kwargs)
return utils.request(method, url, session=self._session, **more)
return utils.http.request(method, url, session=self._session, **more)
def get_default_headers(self):
return {

4
vdirsyncer/storage/http.py
View file

@ -2,8 +2,10 @@
from .base import Item, Storage
from ..exceptions import NotFoundError
from ..utils import expand_path, get_password, request
from ..utils import expand_path
from ..utils.compat import iteritems, text_type, urlparse
from ..utils.http import request
from ..utils.password import get_password
from ..utils.vobject import split_collection
USERAGENT = 'vdirsyncer'

214
vdirsyncer/utils/__init__.py
View file

@ -2,28 +2,14 @@
import os
import sys
import threading
import requests
from .compat import iteritems, urlparse
from .compat import iteritems
from .. import exceptions, log
from ..doubleclick import click, ctx
logger = log.get(__name__)
_missing = object()
try:
import keyring
except ImportError:
keyring = None
password_key_prefix = 'vdirsyncer:'
def expand_path(p):
p = os.path.expanduser(p)
p = os.path.normpath(p)
@ -60,204 +46,6 @@ def uniq(s):
yield x
def get_password(username, resource, _lock=threading.Lock()):
"""tries to access saved password or asks user for it
will try the following in this order:
1. read password from netrc (and only the password, username
in netrc will be ignored)
2. read password from keyring (keyring needs to be installed)
3. read password from the command passed as password_command in the
general config section with username and host as parameters
4a ask user for the password
b save in keyring if installed and user agrees
:param username: user's name on the server
:type username: str/unicode
:param resource: a resource to which the user has access via password,
it will be shortened to just the hostname. It is assumed
that each unique username/hostname combination only ever
uses the same password.
:type resource: str/unicode
:return: password
:rtype: str/unicode
"""
if ctx:
password_cache = ctx.obj.setdefault('passwords', {})
with _lock:
host = urlparse.urlsplit(resource).hostname
for func in (_password_from_command, _password_from_cache,
_password_from_netrc, _password_from_keyring):
password = func(username, host)
if password is not None:
logger.debug('Got password for {} from {}'
.format(username, func.__doc__))
return password
prompt = ('Server password for {} at host {}'.format(username, host))
password = click.prompt(prompt, hide_input=True)
if ctx and func is not _password_from_cache:
password_cache[(username, host)] = password
if keyring is not None and \
click.confirm('Save this password in the keyring?',
default=False):
keyring.set_password(password_key_prefix + host,
username, password)
return password
def _password_from_cache(username, host):
'''internal cache'''
if ctx:
return ctx.obj['passwords'].get((username, host), None)
def _password_from_netrc(username, host):
'''.netrc'''
from netrc import netrc
try:
netrc_user, account, password = \
netrc().authenticators(host) or (None, None, None)
if netrc_user == username:
return password
except IOError:
pass
def _password_from_keyring(username, host):
'''system keyring'''
if keyring is None:
return None
return keyring.get_password(password_key_prefix + host, username)
def _password_from_command(username, host):
'''command'''
import subprocess
if not ctx:
return None
try:
general, _, _ = ctx.obj['config']
command = general['password_command'].split()
except KeyError:
return None
if not command:
return None
command[0] = expand_path(command[0])
try:
stdout = subprocess.check_output(command + [username, host],
universal_newlines=True)
return stdout.strip()
except OSError as e:
logger.warning('Failed to execute command: {}\n{}'.
format(' '.join(command), str(e)))
def _verify_fingerprint_works():
try:
import requests
from pkg_resources import parse_version as ver
return ver(requests.__version__) >= ver('2.4.1')
except Exception:
return False
# https://github.com/shazow/urllib3/pull/444
#
# Without the above pull request, `verify=False` also disables fingerprint
# validation. This is *not* what we want, and it's not possible to replicate
# vdirsyncer's current behavior (verifying fingerprints without verifying
# against CAs) with older versions of urllib3.
#
# We check this here instead of setup.py, because:
# - Python's packaging stuff doesn't check installed versions.
# - The people who don't use `verify_fingerprint` wouldn't care.
VERIFY_FINGERPRINT_WORKS = _verify_fingerprint_works()
del _verify_fingerprint_works
def _install_fingerprint_adapter(session, fingerprint):
prefix = 'https://'
try:
from requests_toolbelt.adapters.fingerprint import \
FingerprintAdapter
except ImportError:
raise RuntimeError('`verify_fingerprint` can only be used with '
'requests-toolbelt versions >= 0.4.0')
if not isinstance(session.adapters[prefix], FingerprintAdapter):
fingerprint_adapter = FingerprintAdapter(fingerprint)
session.mount(prefix, fingerprint_adapter)
def request(method, url, session=None, latin1_fallback=True,
verify_fingerprint=None, **kwargs):
'''
Wrapper method for requests, to ease logging and mocking. Parameters should
be the same as for ``requests.request``, except:
:param session: A requests session object to use.
:param verify_fingerprint: Optional. SHA1 or MD5 fingerprint of the
expected server certificate.
:param latin1_fallback: RFC-2616 specifies the default Content-Type of
text/* to be latin1, which is not always correct, but exactly what
requests is doing. Setting this parameter to False will use charset
autodetection (usually ending up with utf8) instead of plainly falling
back to this silly default. See
https://github.com/kennethreitz/requests/issues/2042
'''
if session is None:
session = requests.Session()
if verify_fingerprint is not None:
if not VERIFY_FINGERPRINT_WORKS:
raise RuntimeError('`verify_fingerprint` can only be used with '
'requests versions >= 2.4.1')
_install_fingerprint_adapter(session, verify_fingerprint)
kwargs['verify'] = False
func = session.request
logger.debug(u'{} {}'.format(method, url))
logger.debug(kwargs.get('headers', {}))
logger.debug(kwargs.get('data', None))
logger.debug('Sending request...')
r = func(method, url, **kwargs)
# See https://github.com/kennethreitz/requests/issues/2042
content_type = r.headers.get('Content-Type', '')
if not latin1_fallback and \
'charset' not in content_type and \
content_type.startswith('text/'):
logger.debug('Removing latin1 fallback')
r.encoding = None
logger.debug(r.status_code)
logger.debug(r.headers)
logger.debug(r.content)
if r.status_code == 412:
raise exceptions.PreconditionFailed(r.reason)
if r.status_code == 404:
raise exceptions.NotFoundError(r.reason)
r.raise_for_status()
return r
def get_etag_from_file(fpath):
'''Get mtime-based etag from a filepath.'''
stat = os.stat(fpath)

100
vdirsyncer/utils/http.py Normal file
View file

@ -0,0 +1,100 @@
# -*- coding: utf-8 -*-
import requests
from .. import exceptions, log
logger = log.get(__name__)
def _verify_fingerprint_works():
try:
from pkg_resources import parse_version as ver
return ver(requests.__version__) >= ver('2.4.1')
except Exception:
return False
# https://github.com/shazow/urllib3/pull/444
#
# Without the above pull request, `verify=False` also disables fingerprint
# validation. This is *not* what we want, and it's not possible to replicate
# vdirsyncer's current behavior (verifying fingerprints without verifying
# against CAs) with older versions of urllib3.
#
# We check this here instead of setup.py, because:
# - Python's packaging stuff doesn't check installed versions.
# - The people who don't use `verify_fingerprint` wouldn't care.
VERIFY_FINGERPRINT_WORKS = _verify_fingerprint_works()
del _verify_fingerprint_works
def _install_fingerprint_adapter(session, fingerprint):
prefix = 'https://'
try:
from requests_toolbelt.adapters.fingerprint import \
FingerprintAdapter
except ImportError:
raise RuntimeError('`verify_fingerprint` can only be used with '
'requests-toolbelt versions >= 0.4.0')
if not isinstance(session.adapters[prefix], FingerprintAdapter):
fingerprint_adapter = FingerprintAdapter(fingerprint)
session.mount(prefix, fingerprint_adapter)
def request(method, url, session=None, latin1_fallback=True,
verify_fingerprint=None, **kwargs):
'''
Wrapper method for requests, to ease logging and mocking. Parameters should
be the same as for ``requests.request``, except:
:param session: A requests session object to use.
:param verify_fingerprint: Optional. SHA1 or MD5 fingerprint of the
expected server certificate.
:param latin1_fallback: RFC-2616 specifies the default Content-Type of
text/* to be latin1, which is not always correct, but exactly what
requests is doing. Setting this parameter to False will use charset
autodetection (usually ending up with utf8) instead of plainly falling
back to this silly default. See
https://github.com/kennethreitz/requests/issues/2042
'''
if session is None:
session = requests.Session()
if verify_fingerprint is not None:
if not VERIFY_FINGERPRINT_WORKS:
raise RuntimeError('`verify_fingerprint` can only be used with '
'requests versions >= 2.4.1')
_install_fingerprint_adapter(session, verify_fingerprint)
kwargs['verify'] = False
func = session.request
logger.debug(u'{} {}'.format(method, url))
logger.debug(kwargs.get('headers', {}))
logger.debug(kwargs.get('data', None))
logger.debug('Sending request...')
r = func(method, url, **kwargs)
# See https://github.com/kennethreitz/requests/issues/2042
content_type = r.headers.get('Content-Type', '')
if not latin1_fallback and \
'charset' not in content_type and \
content_type.startswith('text/'):
logger.debug('Removing latin1 fallback')
r.encoding = None
logger.debug(r.status_code)
logger.debug(r.headers)
logger.debug(r.content)
if r.status_code == 412:
raise exceptions.PreconditionFailed(r.reason)
if r.status_code == 404:
raise exceptions.NotFoundError(r.reason)
r.raise_for_status()
return r

121
vdirsyncer/utils/password.py Normal file
View file

@ -0,0 +1,121 @@
# -*- coding: utf-8 -*-
import threading
import urlparse
from . import expand_path
from .. import log
from ..doubleclick import click, ctx
logger = log.get(__name__)
password_key_prefix = 'vdirsyncer:'
try:
import keyring
except ImportError:
keyring = None
def get_password(username, resource, _lock=threading.Lock()):
"""tries to access saved password or asks user for it
will try the following in this order:
1. read password from netrc (and only the password, username
in netrc will be ignored)
2. read password from keyring (keyring needs to be installed)
3. read password from the command passed as password_command in the
general config section with username and host as parameters
4a ask user for the password
b save in keyring if installed and user agrees
:param username: user's name on the server
:type username: str/unicode
:param resource: a resource to which the user has access via password,
it will be shortened to just the hostname. It is assumed
that each unique username/hostname combination only ever
uses the same password.
:type resource: str/unicode
:return: password
:rtype: str/unicode
"""
if ctx:
password_cache = ctx.obj.setdefault('passwords', {})
with _lock:
host = urlparse.urlsplit(resource).hostname
for func in (_password_from_command, _password_from_cache,
_password_from_netrc, _password_from_keyring):
password = func(username, host)
if password is not None:
logger.debug('Got password for {} from {}'
.format(username, func.__doc__))
return password
prompt = ('Server password for {} at host {}'.format(username, host))
password = click.prompt(prompt, hide_input=True)
if ctx and func is not _password_from_cache:
password_cache[(username, host)] = password
if keyring is not None and \
click.confirm('Save this password in the keyring?',
default=False):
keyring.set_password(password_key_prefix + host,
username, password)
return password
def _password_from_cache(username, host):
'''internal cache'''
if ctx:
return ctx.obj['passwords'].get((username, host), None)
def _password_from_netrc(username, host):
'''.netrc'''
from netrc import netrc
try:
netrc_user, account, password = \
netrc().authenticators(host) or (None, None, None)
if netrc_user == username:
return password
except IOError:
pass
def _password_from_keyring(username, host):
'''system keyring'''
if keyring is None:
return None
return keyring.get_password(password_key_prefix + host, username)
def _password_from_command(username, host):
'''command'''
import subprocess
if not ctx:
return None
try:
general, _, _ = ctx.obj['config']
command = general['password_command'].split()
except KeyError:
return None
if not command:
return None
command[0] = expand_path(command[0])
try:
stdout = subprocess.check_output(command + [username, host],
universal_newlines=True)
return stdout.strip()
except OSError as e:
logger.warning('Failed to execute command: {}\n{}'.
format(' '.join(command), str(e)))