diff options
| -rw-r--r-- | CHANGES | 5 | ||||
| -rw-r--r-- | docs/lib/passlib.apache.rst | 33 | ||||
| -rw-r--r-- | passlib/apache.py | 1142 | ||||
| -rw-r--r-- | passlib/handlers/digests.py | 62 | ||||
| -rw-r--r-- | passlib/registry.py | 1 | ||||
| -rw-r--r-- | passlib/tests/test_apache.py | 347 | ||||
| -rw-r--r-- | passlib/tests/test_handlers.py | 31 | ||||
| -rw-r--r-- | passlib/utils/__init__.py | 39 |
8 files changed, 1163 insertions, 497 deletions
@@ -143,6 +143,11 @@ Release History Other + * The api for the :mod:`passlib.apache` module has been updated + to add more flexibility, and to fix some ambiguous method + and keyword names. The old names are still supported, but deprecated, + and will be removed in Passlib 1.8. + * Handle platform-specific error strings returned by :func:`!crypt.crypt`. * Passlib is now source-compatible with Python 2.5+ and Python 3, diff --git a/docs/lib/passlib.apache.rst b/docs/lib/passlib.apache.rst index 731baed..8649e62 100644 --- a/docs/lib/passlib.apache.rst +++ b/docs/lib/passlib.apache.rst @@ -8,6 +8,13 @@ This module provides utilities for reading and writing Apache's htpasswd and htdigest files; though the use of two helper classes. +.. versionchanged:: 1.6 + The api for this module was updated to be more flexible, + and to have (hopefully) less confusing method names. + The old method and keyword names are supported but deprecated, and + will be removed in Passlib 1.8. + No more backwards-incompatible changes are currently planned. + .. index:: apache; htpasswd Htpasswd Files @@ -17,34 +24,34 @@ A quick summary of it's usage:: >>> from passlib.apache import HtpasswdFile - >>> #when creating a new file, set to autoload=False, add entries, and save. - >>> ht = HtpasswdFile("test.htpasswd", autoload=False) - >>> ht.update("someuser", "really secret password") + >>> # when creating a new file, set to new=True, add entries, and save. + >>> ht = HtpasswdFile("test.htpasswd", new=True) + >>> ht.set_password("someuser", "really secret password") >>> ht.save() - >>> #loading an existing file to update a password + >>> # loading an existing file to update a password >>> ht = HtpasswdFile("test.htpasswd") - >>> ht.update("someuser", "new secret password") + >>> ht.set_password("someuser", "new secret password") >>> ht.save() - >>> #examining file, verifying user's password + >>> # examining file, verifying user's password >>> ht = HtpasswdFile("test.htpasswd") >>> ht.users() [ "someuser" ] - >>> ht.verify("someuser", "wrong password") + >>> ht.check_password("someuser", "wrong password") False - >>> ht.verify("someuser", "new secret password") + >>> ht.check_password("someuser", "new secret password") True - >>> #making in-memory changes and exporting to string + >>> # making in-memory changes and exporting to string >>> ht = HtpasswdFile() - >>> ht.update("someuser", "mypass") - >>> ht.update("someuser", "anotherpass") + >>> ht.set_password("someuser", "mypass") + >>> ht.set_password("someuser", "anotherpass") >>> print ht.to_string() someuser:$apr1$T4f7D9ly$EobZDROnHblCNPCtrgh5i/ anotheruser:$apr1$vBdPWvh1$GrhfbyGvN/7HalW5cS9XB1 -.. autoclass:: HtpasswdFile(path, default=None, autoload=True) +.. autoclass:: HtpasswdFile(path=None, new=False, autosave=False, ...) .. index:: apache; htdigest @@ -53,7 +60,7 @@ Htdigest Files The :class:`!HtdigestFile` class allows management of htdigest files in a similar fashion to :class:`HtpasswdFile`. -.. autoclass:: HtdigestFile(path, autoload=True) +.. autoclass:: HtdigestFile(path, default_realm=None, new=False, autosave=False, ...) .. rubric:: Footnotes diff --git a/passlib/apache.py b/passlib/apache.py index 05f4b68..e7c3e25 100644 --- a/passlib/apache.py +++ b/passlib/apache.py @@ -11,390 +11,722 @@ import sys #site #libs from passlib.context import CryptContext -from passlib.utils import consteq, render_bytes -from passlib.utils.compat import b, bytes, join_bytes, lmap, str_to_bascii, u, unicode +from passlib.exc import ExpectedStringError +from passlib.hash import htdigest +from passlib.utils import consteq, render_bytes, to_bytes, deprecated_method +from passlib.utils.compat import b, bytes, join_bytes, str_to_bascii, u, \ + unicode, BytesIO, iteritems, imap, PY3 #pkg #local __all__ = [ + 'HtpasswdFile', + 'HtdigestFile', ] -BCOLON = b(":") +#========================================================= +# constants & support +#========================================================= +_UNSET = object() + +_BCOLON = b(":") + +# byte values that aren't allowed in fields. +_INVALID_FIELD_CHARS = b(":\n\r\t\x00") + +# helpers to detect non-ascii codecs +_ASCII_TEST_BYTES = b("\x00\n aA:#!\x7f") +_ASCII_TEST_UNICODE = _ASCII_TEST_BYTES.decode("ascii") + +def is_ascii_codec(codec): + "test if codec is 7-bit ascii safe (e.g. latin-1, utf-8; but not utf-16)" + return _ASCII_TEST_UNICODE.encode(codec) == _ASCII_TEST_BYTES #========================================================= -#common helpers +# backport of OrderedDict for PY2.5 #========================================================= -DEFAULT_ENCODING = "utf-8" if sys.version_info >= (3,0) else None +try: + from collections import OrderedDict +except ImportError: + # Python 2.5 + class OrderedDict(dict): + """hacked OrderedDict replacement. + + NOTE: this doesn't provide a full OrderedDict implementation, + just the minimum needed by the Htpasswd internals. + """ + def __init__(self): + self._keys = [] + + def __iter__(self): + return iter(self._keys) + def __setitem__(self, key, value): + if key not in self: + self._keys.append(key) + super(OrderedDict, self).__setitem__(key, value) + + def __delitem__(self, key): + super(OrderedDict, self).__delitem__(key) + self._keys.remove(key) + + def iteritems(self): + return ((key, self[key]) for key in self) + + # these aren't used or implemented, so disabling them for safety. + update = pop = popitem = clear = keys = iterkeys = None + +#========================================================= +#common helpers +#========================================================= class _CommonFile(object): - "helper for HtpasswdFile / HtdigestFile" - - #XXX: would like to add 'path' keyword to load() / save(), - # but that makes .mtime somewhat meaningless. - # to simplify things, should probably deprecate mtime & force=False - # options. - #XXX: would also like to make _load_string available via public interface, - # such as via 'content' keyword in load() method. - # in short, need to clean up the htpasswd api a little bit in 1.6. - # keeping _load_string private for now, cause just using it for UTing. - - #NOTE: 'path' is a property instead of attr, - # so that .mtime is wiped whenever path is changed. - _path = None - def _get_path(self): - return self._path - def _set_path(self, path): - if path != self._path: - self.mtime = 0 - self._path = path - path = property(_get_path, _set_path) + """common framework for HtpasswdFile & HtdigestFile""" + #======================================================================= + # instance attrs + #======================================================================= + + # charset encoding used by file (defaults to utf-8) + encoding = None + + # whether users() and other public methods should return unicode or bytes? + # (defaults to False under PY2, True under PY3) + return_unicode = None + + # if bound to local file, these will be set. + _path = None # local file path + _mtime = None # mtime when last loaded, or 0 + + # if true, automatically save to local file after changes are made. + autosave = False + + # ordered dict mapping key -> value for all records in database. + # (e.g. user => hash for Htpasswd) + _records = None + + #======================================================================= + # alt constuctors + #======================================================================= + @classmethod + def from_string(cls, data, **kwds): + """create new object from raw string. + + :arg data: + unicode or bytes string to load + + :param \*\*kwds: + all other keywords are the same as in the class constructor + """ + if 'path' in kwds: + raise TypeError("'path' not accepted by from_string()") + self = cls(**kwds) + self.load_string(data) + return self @classmethod - def _from_string(cls, content, **kwds): - #NOTE: not public yet, just using it for unit tests. + def from_path(cls, path, **kwds): + """create new object from file, without binding object to file. + + :arg path: + local filepath to load from + + :param \*\*kwds: + all other keywords are the same as in the class constructor + """ self = cls(**kwds) - self._load_string(content) + self.load(path) return self - def __init__(self, path=None, autoload=True, - encoding=DEFAULT_ENCODING, + #======================================================================= + # init + #======================================================================= + def __init__(self, path=None, new=False, autoload=True, autosave=False, + encoding="utf-8", return_unicode=PY3, ): - if encoding and u(":\n").encode(encoding) != b(":\n"): - #rest of file assumes ascii bytes, and uses ":" as separator. + # set encoding + if not encoding: + warn("``encoding=None`` is deprecated as of Passlib 1.6, " + "and will cause a ValueError in Passlib 1.8, " + "use ``return_unicode=False`` instead.", + DeprecationWarning, stacklevel=2) + encoding = "utf-8" + return_unicode = False + elif not is_ascii_codec(encoding): + # htpasswd/htdigest files assumes 1-byte chars, and use ":" separator, + # so only ascii-compatible encodings are allowed. raise ValueError("encoding must be 7-bit ascii compatible") self.encoding = encoding - self.path = path - ##if autoload == "exists": - ## autoload = bool(path and os.path.exists(path)) - if autoload and path: + + # set other attrs + self.return_unicode = return_unicode + self.autosave = autosave + self._path = path + self._mtime = 0 + + # init db + if not autoload: + warn("``autoload=False`` is deprecated as of Passlib 1.6, " + "and will be removed in Passlib 1.8, use ``new=True`` instead", + DeprecationWarning, stacklevel=2) + new = True + if path and not new: self.load() - ##elif raw: - ## self._load_lines(raw.split("\n")) else: - self._entry_order = [] - self._entry_map = {} - - def _load_string(self, content): - """UT helper for loading from string - - to be improved/made public in later release. - + self._records = OrderedDict() + + def __repr__(self): + tail = '' + if self.autosave: + tail += ' autosave=True' + if self._path: + tail += ' path=%r' % self._path + if self.encoding != "utf-8": + tail += ' encoding=%r' % self.encoding + return "<%s 0x%0x%s>" % (self.__class__.__name__, id(self), tail) + + # NOTE: ``path`` is a property so that ``_mtime`` is wiped when it's set. + def _get_path(self): + return self._path + def _set_path(self, value): + if value != self._path: + self._mtime = 0 + self._path = value + path = property(_get_path, _set_path) - :param content: - if specified, should be a bytes object. - passwords will be loaded directly from this string, - and any files will be ignored. - """ - if isinstance(content, unicode): - content = content.encode(self.encoding or 'utf-8') - self.mtime = 0 - #XXX: replace this with iterator? - lines = content.splitlines() - self._load_lines(lines) + @property + def mtime(self): + "modify time when last loaded (if bound to a local file)" + return self._mtime + + #======================================================================= + # loading + #======================================================================= + def load_if_changed(self): + """Reload from ``self.path`` only if file has changed since last load""" + if not self._path: + raise RuntimeError("%r is not bound to a local file" % self) + if self._mtime and self._mtime == os.path.getmtime(self._path): + return False + self.load() return True - def load(self, force=True): - """load entries from file + def load(self, path=None, force=True): + """Load state from local file. + If no path is specified, attempts to load from ``self.path``. - :param force: - if ``True`` (the default), always loads state from file. - if ``False``, only loads state if file has been modified since last load. + :arg path: local file to load from - :raises IOError: if file not found + :param force: + if ``force=False``, only load from ``self.path`` if file + has changed since last load. - :returns: ``False`` if ``force=False`` and no load performed; otherwise ``True``. + .. deprecated:: 1.6 + This keyword will be removed in Passlib 1.8; + Applications should use :meth:`load_if_changed` instead. """ - path = self.path - if not path: - raise RuntimeError("no load path specified") - if not force and self.mtime and self.mtime == os.path.getmtime(path): - return False - with open(path, "rb") as fh: - self.mtime = os.path.getmtime(path) - self._load_lines(fh) + if path is not None: + with open(path, "rb") as fh: + self._mtime = 0 + self._load_lines(fh) + elif not force: + warn("%(name)s.load(force=False) is deprecated as of Passlib 1.6," + "and will be removed in Passlib 1.8; " + "use %(name)s.load_if_changed() instead." % + self.__class__.__name__, + DeprecationWarning, stacklevel=2) + return self.load_if_changed() + elif self._path: + with open(self._path, "rb") as fh: + self._mtime = os.path.getmtime(self._path) + self._load_lines(fh) + else: + raise RuntimeError("%s().path is not set, an explicit path is required" % + self.__class__.__name__) return True + def load_string(self, data): + "Load state from unicode or bytes string, replacing current state" + data = to_bytes(data, self.encoding, "data") + self._mtime = 0 + self._load_lines(BytesIO(data)) + def _load_lines(self, lines): - pl = self._parse_line - entry_order = self._entry_order = [] - entry_map = self._entry_map = {} - for line in lines: - #XXX: found mention that "#" comment lines may be supported by htpasswd, - # should verify this. - key, value = pl(line) - if key in entry_map: - #XXX: should we use data from first entry, or last entry? - # going w/ first entry for now. - continue - entry_order.append(key) - entry_map[key] = value - - #subclass: _parse_line(line) -> (key, hash) + "load from sequence of lists" + # XXX: found reference that "#" comment lines may be supported by + # htpasswd, should verify this, and figure out how to handle them. + # if true, this would also affect what can be stored in user field. + # XXX: if multiple entries for a key, should we use the first one + # or the last one? going w/ first entry for now. + # XXX: how should this behave if parsing fails? currently + # it will contain everything that was loaded up to error. + # could clear / restore old state instead. + parse = self._parse_record + records = self._records = OrderedDict() + for idx, line in enumerate(lines): + key, value = parse(line, idx+1) + if key not in records: + records[key] = value + + def _parse_record(cls, record, lineno): + "parse line of file into (key, value) pair" + raise NotImplementedError("should be implemented in subclass") + + #======================================================================= + # saving + #======================================================================= + def _autosave(self): + "subclass helper to call save() after any changes" + if self.autosave and self._path: + self.save() + + def save(self, path=None): + """Save current state to file. + If no path is specified, attempts to save to ``self.path``. + """ + if path is not None: + with open(path, "wb") as fh: + fh.writelines(self._iter_lines()) + elif self._path: + self.save(self._path) + self._mtime = os.path.getmtime(self._path) + else: + raise RuntimeError("%s().path is not set, cannot autosave" % + self.__class__.__name__) + + def to_string(self): + "Export current state as a string of bytes" + return join_bytes(self._iter_lines()) def _iter_lines(self): "iterator yielding lines of database" - rl = self._render_line - entry_order = self._entry_order - entry_map = self._entry_map - assert len(entry_order) == len(entry_map), "internal error in entry list" - return (rl(key, entry_map[key]) for key in entry_order) - - def save(self): - "save entries to file" - if not self.path: - raise RuntimeError("no save path specified") - with open(self.path, "wb") as fh: - fh.writelines(self._iter_lines()) - self.mtime = os.path.getmtime(self.path) + return (self._render_record(key,value) for key,value in iteritems(self._records)) - def to_string(self): - "export whole database as a byte string" - return join_bytes(self._iter_lines()) + def _render_record(cls, key, value): + "given key/value pair, encode as line of file" + raise NotImplementedError("should be implemented in subclass") - #subclass: _render_line(entry) -> line + #======================================================================= + # field encoding + #======================================================================= + def _encode_user(self, user): + "user-specific wrapper for _encode_field()" + return self._encode_field(user, "user") - def _update_key(self, key, value): - entry_map = self._entry_map - if key in entry_map: - entry_map[key] = value - return True - else: - self._entry_order.append(key) - entry_map[key] = value - return False + def _encode_realm(self, realm): + "realm-specific wrapper for _encode_field()" + return self._encode_field(realm, "realm") - def _delete_key(self, key): - entry_map = self._entry_map - if key in entry_map: - del entry_map[key] - self._entry_order.remove(key) - return True - else: - return False + def _encode_field(self, value, errname="field"): + """convert field to internal representation. - invalid_chars = b(":\n\r\t\x00") - - def _norm_user(self, user): - "encode user to bytes, validate against format requirements" - return self._norm_ident(user, errname="user") - - def _norm_realm(self, realm): - "encode realm to bytes, validate against format requirements" - return self._norm_ident(realm, errname="realm") - - def _norm_ident(self, ident, errname="user/realm"): - ident = self._encode_ident(ident, errname) - if len(ident) > 255: - raise ValueError("%s must be at most 255 characters: %r" % (errname, ident)) - if any(c in self.invalid_chars for c in ident): - raise ValueError("%s contains invalid characters: %r" % (errname, ident,)) - return ident - - def _encode_ident(self, ident, errname="user/realm"): - "ensure identifier is bytes encoded using specified encoding, or rejected" - encoding = self.encoding - if encoding: - if isinstance(ident, unicode): - return ident.encode(encoding) - raise TypeError("%s must be unicode, not %s" % - (errname, type(ident))) - else: - if isinstance(ident, bytes): - return ident - raise TypeError("%s must be bytes, not %s" % - (errname, type(ident))) - - def _decode_ident(self, ident, errname="user/realm"): - "decode an identifier (if encoding is specified, else return encoded bytes)" - assert isinstance(ident, bytes) - encoding = self.encoding - if encoding: - return ident.decode(encoding) + internal representation is always bytes. byte strings are left as-is, + unicode strings encoding using file's default encoding (or ``utf-8`` + if no encoding has been specified). + + :raises UnicodeEncodeError: + if unicode value cannot be encoded using default encoding. + + :raises ValueError: + if resulting byte string contains a forbidden character, + or is too long (>255 bytes). + + :returns: + encoded identifer as bytes + """ + if isinstance(value, unicode): + value = value.encode(self.encoding) + elif not isinstance(value, bytes): + raise ExpectedStringError(value, errname) + if len(value) > 255: + raise ValueError("%s must be at most 255 characters: %r" % + (errname, value)) + if any(c in _INVALID_FIELD_CHARS for c in value): + raise ValueError("%s contains invalid characters: %r" % + (errname, value,)) + return value + + def _decode_field(self, value): + """decode field from internal representation to format + returns by users() method, etc. + + :raises UnicodeDecodeError: + if unicode value cannot be decoded using default encoding. + (usually indicates wrong encoding set for file). + + :returns: + field as unicode or bytes, as appropriate. + """ + assert isinstance(value, bytes), "expected value to be bytes" + if self.return_unicode: + return value.decode(self.encoding) else: - return ident + return value - #FIXME: htpasswd doc sez passwords limited to 255 chars under Windows & MPE, - # longer ones are truncated. may be side-effect of those platforms - # supporting plaintext. we don't currently check for this. + # FIXME: htpasswd doc says passwords limited to 255 chars under Windows & MPE, + # and that longer ones are truncated. this may be side-effect of those + # platforms supporting the 'plaintext' scheme. these classes don't currently + # check for this. + + #======================================================================= + # eoc + #======================================================================= #========================================================= #htpasswd editing #========================================================= -#FIXME: apr_md5_crypt technically the default only for windows, netware and tpf. -#TODO: find out if htpasswd's "crypt" mode is crypt *call* or just des_crypt implementation. + +# FIXME: apr_md5_crypt technically the default only for windows, netware and tpf. +# TODO: find out if htpasswd's "crypt" mode is crypt *call* or just des_crypt implementation. htpasswd_context = CryptContext([ - "apr_md5_crypt", #man page notes supported everywhere, default on Windows, Netware, TPF - "des_crypt", #man page notes server does NOT support this on Windows, Netware, TPF - "ldap_sha1", #man page notes only for transitioning <-> ldap + "apr_md5_crypt", # man page notes supported everywhere, default on Windows, Netware, TPF + "des_crypt", # man page notes server does NOT support this on Windows, Netware, TPF + "ldap_sha1", # man page notes only for transitioning <-> ldap "plaintext" # man page notes server ONLY supports this on Windows, Netware, TPF ]) class HtpasswdFile(_CommonFile): """class for reading & writing Htpasswd files. - :arg path: path to htpasswd file to load from / save to (required) + The class constructor accepts the following arguments: - :param default: - optionally specify default scheme to use when encoding new passwords. + :type path: filepath + :param path: - Must be one of ``None``, ``"apr_md5_crypt"``, ``"des_crypt"``, ``"ldap_sha1"``, ``"plaintext"``. + Specifies path to htpasswd file, use to implicitly load from and save to. - If no value is specified, this class currently uses ``apr_md5_crypt`` when creating new passwords. + This class has two modes of operation: - :param autoload: - if ``True`` (the default), :meth:`load` will be automatically called - by constructor. + 1. It can be "bound" to a local file by passing a ``path`` to the class + constructor. In this case it will load the contents of the file when + created, and the :meth:`load` and :meth:`save` methods will automatically + load from and save to that file if they are called without arguments. - Set to ``False`` to disable automatic loading (primarily used when - creating new htdigest file). + 2. Alternately, it can exist as an independant object, in which case + :meth:`load` and :meth:`save` will require an explicit path to be + provided whenever they are called. As well, ``autosave`` behavior + will not be available. - :param encoding: - optionally specify encoding used for usernames. + This feature is new in Passlib 1.6, and is the default if no + ``path`` value is provided to the constructor. + + This is exposed as a readonly instance attribute. + + :type new: bool + :param new: + + Normally, if *path* is specified, :class:`HtpasswdFile` will + immediately load the contents of the file. However, when creating + a new htpasswd file, applications can set ``new=True`` so that + the existing file (if any) will not be loaded. + + .. versionchanged:: 1.6 + This feature was previously enabled by setting ``autoload=False``. + That alias has been deprecated, and will be removed in Passlib 1.8 - if set to ``None``, - user names must be specified as bytes, - and will be returned as bytes. + :type autosave: bool + :param autosave: + + Normally, any changes made to an :class:`HtpasswdFile` instance + will not be saved until :meth:`save` is explicitly called. However, + if ``autosave=True`` is specified, any changes made will be + saved to disk immediately (assuming *path* has been set). + + This is exposed as a writeable instance attribute. + + :type encoding: str + :param encoding: - if set to an encoding, - user names must be specified as unicode, - and will be returned as unicode. - when stored, then will use the specified encoding. + Optionally specify character encoding used to read/write file + and hash passwords. Defaults to ``utf-8``, though ``latin-1`` + is the only other commonly encountered encoding. - for backwards compatibility with passlib 1.4, - this defaults to ``None`` under Python 2, - and ``utf-8`` under Python 3. + This is exposed as a readonly instance attribute. - .. note:: + :type default_scheme: str + :param default_scheme: + Optionally specify default scheme to use when encoding new passwords. + Must be one of ``"apr_md5_crypt"``, ``"des_crypt"``, ``"ldap_sha1"``, + ``"plaintext"``. It defaults to ``"apr_md5_crypt"``. - this is not the encoding for the entire file, - just for the usernames within the file. - this must be an encoding which is compatible - with 7-bit ascii (which is used by rest of file). + .. versionchanged:: 1.6 + This keyword was previously named ``default``. That alias + has been deprecated, and will be removed in Passlib 1.8. + :type context: :class:`~passlib.context.CryptContext` :param context: - :class:`~passlib.context.CryptContext` instance used to handle - hashes in this file. + :class:`!CryptContext` instance used to encrypt + and verify the hashes found in the htpasswd file. + The default value is a pre-built context which supports all + of the hashes officially allowed in an htpasswd file. + + This is exposed as a readonly instance attribute. .. warning:: - this should usually be left at the default, - though it can be overridden to implement non-standard hashes - within the htpasswd file. + This option is useful to add support for non-standard hash + formats to an htpasswd file. However, the resulting file + will probably not be usuable by another application, + particularly Apache itself. Loading & Saving ================ .. automethod:: load + .. automethod:: load_if_changed + .. automethod:: load_string .. automethod:: save .. automethod:: to_string Inspection ================ .. automethod:: users - .. automethod:: verify + .. automethod:: check_password + .. automethod:: get_hash Modification ================ - .. automethod:: update + .. automethod:: set_password .. automethod:: delete - .. note:: + Alternate Constructors + ====================== + .. automethod:: from_string - All of the methods in this class enforce some data validation - on the ``user`` parameter: - they will raise a :exc:`ValueError` if the string - contains one of the forbidden characters ``:\\r\\n\\t\\x00``, + Errors + ====== + :raises ValueError: + All of the methods in this class will raise a :exc:`ValueError` if + any user name contains a forbidden character (one of ``:\\r\\n\\t\\x00``), or is longer than 255 characters. """ - def __init__(self, path=None, default=None, context=htpasswd_context, **kwds): + #========================================================= + # instance attrs + #========================================================= + + # NOTE: _records map stores <user> for the key, and <hash> for the value, + # both in bytes which use self.encoding + + #========================================================= + # init & serialization + #========================================================= + def __init__(self, path=None, default_scheme=None, context=htpasswd_context, + **kwds): + if 'default' in kwds: + warn("``default`` is deprecated as of Passlib 1.6, " + "and will be removed in Passlib 1.8, it has been renamed " + "to ``default_scheem``.", + DeprecationWarning, stacklevel=2) + default_scheme = kwds.pop("default") + if default_scheme: + context = context.replace(default=default_scheme) self.context = context - if default: - self.context = self.context.replace(default=default) super(HtpasswdFile, self).__init__(path, **kwds) - def _parse_line(self, line): - #should be user, hash - return line.rstrip().split(BCOLON) + def _parse_record(self, record, lineno): + # NOTE: should return (user, hash) tuple + result = record.rstrip().split(_BCOLON) + if len(result) != 2: + raise ValueError("malformed htpasswd file (error reading line %d)" + % lineno) + return result - def _render_line(self, user, hash): + def _render_record(self, user, hash): return render_bytes("%s:%s\n", user, hash) + #========================================================= + # public methods + #========================================================= + def users(self): - "return list of all users in file" - return lmap(self._decode_ident, self._entry_order) + "Return list of all users in database" + return [self._decode_field(user) for user in self._records] - def update(self, user, password): - """update password for user; adds user if needed. + ##def has_user(self, user): + ## "check whether entry is present for user" + ## return self._encode_user(user) in self._records + + ##def rename(self, old, new): + ## """rename user account""" + ## old = self._encode_user(old) + ## new = self._encode_user(new) + ## hash = self._records.pop(old) + ## self._records[new] = hash + ## self._autosave() + + def set_password(self, user, password): + """Set password for user; adds user if needed. - :returns: ``True`` if existing user was updated, ``False`` if user added. + :returns: + * ``True`` if existing user was updated. + * ``False`` if user account was added. + + .. versionchanged:: 1.6 + This method was previously called ``update``, it was renamed + to prevent ambiguity with the dictionary method. + The old alias is deprecated, and will be removed in Passlib 1.8. """ - user = self._norm_user(user) + user = self._encode_user(user) hash = self.context.encrypt(password) - return self._update_key(user, hash) + if PY3: + hash = hash.encode(self.encoding) + existing = (user in self._records) + self._records[user] = hash + self._autosave() + return existing + + @deprecated_method(deprecated="1.6", removed="1.8", + replacement="set_password") + def update(self, user, password): + "set password for user" + return self.set_password(user, password) + + def get_hash(self, user): + """Return hash stored for user, or ``None`` if user not found. + .. versionchanged:: 1.6 + This method was previously named ``find``, it was renamed + for clarity. The old name is deprecated, and will be removed + in Passlib 1.8. + """ + try: + return self._records[self._encode_user(user)] + except KeyError: + return None + + @deprecated_method(deprecated="1.6", removed="1.8", + replacement="get_hash") + def find(self, user): + "return hash for user" + return self.get_hash(user) + + # XXX: rename to something more explicit, like delete_user()? def delete(self, user): - """delete user's entry. + """Delete user's entry. - :returns: ``True`` if user deleted, ``False`` if user not found. + :returns: + * ``True`` if user deleted. + * ``False`` if user not found. """ - user = self._norm_user(user) - return self._delete_key(user) + try: + del self._records[self._encode_user(user)] + except KeyError: + return False + self._autosave() + return True - def verify(self, user, password): - """verify password for specified user. + def check_password(self, user, password): + """Verify password for specified user. :returns: - * ``None`` if user not found - * ``False`` if password does not match - * ``True`` if password matches. + * ``None`` if user not found. + * ``False`` if user found, but password does not match. + * ``True`` if user found and password matches. + + .. versionchanged:: 1.6 + This method was previously called ``verify``, it was renamed + to prevent ambiguity with the :class:`!CryptContext` method. + The old alias is deprecated, and will be removed in Passlib 1.8. """ - user = self._norm_user(user) - hash = self._entry_map.get(user) + user = self._encode_user(user) + hash = self._records.get(user) if hash is None: return None - else: - return self.context.verify(password, hash) - #TODO: support migration from deprecated hashes + if isinstance(password, unicode): + # NOTE: encoding password to match file, making the assumption + # that server will use same encoding to hash the password. + password = password.encode(self.encoding) + ok, new_hash = self.context.verify_and_update(password, hash) + if ok and new_hash is not None: + # rehash user's password if old hash was deprecated + self._records[user] = new_hash + self._autosave() + return ok + + @deprecated_method(deprecated="1.6", removed="1.8", + replacement="check_password") + def verify(self, user, password): + "verify password for user" + return self.check_password(user, password) + + #========================================================= + # eoc + #========================================================= #========================================================= #htdigest editing #========================================================= class HtdigestFile(_CommonFile): - """class for reading & writing Htdigest files + """class for reading & writing Htdigest files. - :arg path: path to htpasswd file to load from / save to (required) + The class constructor accepts the following arguments: - :param autoload: - if ``True`` (the default), :meth:`load` will be automatically called - by constructor. + :type path: filepath + :param path: - Set to ``False`` to disable automatic loading (primarily used when - creating new htdigest file). + Specifies path to htdigest file, use to implicitly load from and save to. - :param encoding: - optionally specify encoding used for usernames / realms. + This class has two modes of operation: + + 1. It can be "bound" to a local file by passing a ``path`` to the class + constructor. In this case it will load the contents of the file when + created, and the :meth:`load` and :meth:`save` methods will automatically + load from and save to that file if they are called without arguments. + + 2. Alternately, it can exist as an independant object, in which case + :meth:`load` and :meth:`save` will require an explicit path to be + provided whenever they are called. As well, ``autosave`` behavior + will not be available. + + This feature is new in Passlib 1.6, and is the default if no + ``path`` value is provided to the constructor. + + This is exposed as a readonly instance attribute. + + :type default_realm: str + :param default_realm: - if set to ``None``, - user names & realms must be specified as bytes, - and will be returned as bytes. + If ``default_realm`` is set, all the :class:`HtdigestFile` + methods that require a realm will use this value if one is not + provided explicitly. If unset, they will raise an error stating + that an explicit realm is required. - if set to an encoding, - user names & realms must be specified as unicode, - and will be returned as unicode. - when stored, then will use the specified encoding. + This is exposed as a writeable instance attribute. - for backwards compatibility with passlib 1.4, - this defaults to ``None`` under Python 2, - and ``utf-8`` under Python 3. + .. versionadded:: 1.6 - .. note:: + :type new: bool + :param new: - this is not the encoding for the entire file, - just for the usernames & realms within the file. - this must be an encoding which is compatible - with 7-bit ascii (which is used by rest of file). + Normally, if *path* is specified, :class:`HtdigestFile` will + immediately load the contents of the file. However, when creating + a new htpasswd file, applications can set ``new=True`` so that + the existing file (if any) will not be loaded. + + .. versionchanged:: 1.6 + This feature was previously enabled by setting ``autoload=False``. + That alias has been deprecated, and will be removed in Passlib 1.8 + + :type autosave: bool + :param autosave: + + Normally, any changes made to an :class:`HtdigestFile` instance + will not be saved until :meth:`save` is explicitly called. However, + if ``autosave=True`` is specified, any changes made will be + saved to disk immediately (assuming *path* has been set). + + This is exposed as a writeable instance attribute. + + :type encoding: str + :param encoding: + + Optionally specify character encoding used to read/write file + and hash passwords. Defaults to ``utf-8``, though ``latin-1`` + is the only other commonly encountered encoding. + + This is exposed as a readonly instance attribute. Loading & Saving ================ .. automethod:: load + .. automethod:: load_if_changed + .. automethod:: load_string .. automethod:: save .. automethod:: to_string @@ -402,130 +734,242 @@ class HtdigestFile(_CommonFile): ========== .. automethod:: realms .. automethod:: users - .. automethod:: find - .. automethod:: verify + .. automethod:: check_password(user[, realm], password) + .. automethod:: get_hash Modification ============ - .. automethod:: update + .. automethod:: set_password(user[, realm], password) .. automethod:: delete .. automethod:: delete_realm - .. note:: + Alternate Constructors + ====================== + .. automethod:: from_string - All of the methods in this class enforce some data validation - on the ``user`` and ``realm`` parameters: - they will raise a :exc:`ValueError` if either string - contains one of the forbidden characters ``:\\r\\n\\t\\x00``, + Errors + ====== + :raises ValueError: + All of the methods in this class will raise a :exc:`ValueError` if + any user name or realm contains a forbidden character (one of ``:\\r\\n\\t\\x00``), or is longer than 255 characters. - """ - #XXX: don't want password encoding to change if user account encoding does. - # but also *can't* use unicode itself. setting this to utf-8 for now, - # until it causes problems - in which case stopgap of setting this attr - # per-instance can be used. - password_encoding = "utf-8" + #========================================================= + # instance attrs + #========================================================= + + # NOTE: _records map stores (<user>,<realm>) for the key, + # and <hash> as the value, all as <self.encoding> bytes. + + # NOTE: unlike htpasswd, this class doesn't use a CryptContext, + # as only one hash format is supported: htdigest. + + # optionally specify default realm that will be used if none + # is provided to a method call. otherwise realm is always required. + default_realm = None + + #========================================================= + # init & serialization + #========================================================= + def __init__(self, path=None, default_realm=None, **kwds): + self.default_realm = default_realm + super(HtdigestFile, self).__init__(path, **kwds) + + def _parse_record(self, record, lineno): + result = record.rstrip().split(_BCOLON) + if len(result) != 3: + raise ValueError("malformed htdigest file (error reading line %d)" + % lineno) + user, realm, hash = result + return (user, realm), hash - #XXX: provide rename() & rename_realm() ? + def _render_record(self, key, hash): + user, realm = key + return render_bytes("%s:%s:%s\n", user, realm, hash) - def _parse_line(self, line): - user, realm, hash = line.rstrip().split(BCOLON) - return (user, realm), hash + def _encode_realm(self, realm): + # override default _encode_realm to fill in default realm field + if realm is None: + realm = self.default_realm + if realm is None: + raise TypeError("you must specify a realm explicitly, " + "or set the default_realm attribute") + return self._encode_field(realm, "realm") - def _render_line(self, key, hash): - return render_bytes("%s:%s:%s\n", key[0], key[1], hash) - - #TODO: would frontend to calc_digest be useful? - ##def encrypt(self, password, user, realm): - ## user = self._norm_user(user) - ## realm = self._norm_realm(realm) - ## hash = self._calc_digest(user, realm, password) - ## if self.encoding: - ## #decode hash if in unicode mode - ## hash = hash.decode("ascii") - ## return hash - - def _calc_digest(self, user, realm, password): - "helper to calculate digest" - if isinstance(password, unicode): - password = password.encode(self.password_encoding) - #NOTE: encode('ascii') is noop under py2, required under py3 - return str_to_bascii(md5(render_bytes("%s:%s:%s", user, realm, password)).hexdigest()) + #========================================================= + # public methods + #========================================================= def realms(self): - "return all realms listed in file" - return lmap(self._decode_ident, - set(key[1] for key in self._entry_order)) + """Return list of all realms in database""" + realms = set(key[1] for key in self._records) + return [self._decode_field(realm) for realm in realms] - def users(self, realm): - "return list of all users within specified realm" - realm = self._norm_realm(realm) - return lmap(self._decode_ident, - (key[0] for key in self._entry_order if key[1] == realm)) + def users(self, realm=None): + """Return list of all users in specified realm. + + * uses ``self.default_realm`` if no realm explicitly provided. + * returns empty list if realm not found. + """ + realm = self._encode_realm(realm) + return [self._decode_field(key[0]) for key in self._records + if key[1] == realm] + + ##def has_user(self, user, realm=None): + ## "check if user+realm combination exists" + ## user = self._encode_user(user) + ## realm = self._encode_realm(realm) + ## return (user,realm) in self._records + + ##def rename_realm(self, old, new): + ## """rename all accounts in realm""" + ## old = self._encode_realm(old) + ## new = self._encode_realm(new) + ## keys = [key for key in self._records if key[1] == old] + ## for key in keys: + ## hash = self._records.pop(key) + ## self._records[key[0],new] = hash + ## self._autosave() + ## return len(keys) + + ##def rename(self, old, new, realm=None): + ## """rename user account""" + ## old = self._encode_user(old) + ## new = self._encode_user(new) + ## realm = self._encode_realm(realm) + ## hash = self._records.pop((old,realm)) + ## self._records[new,realm] = hash + ## self._autosave() + + def set_password(self, user, realm=None, password=_UNSET): + """Set password for user; adds user & realm if needed. + + If ``self.default_realm`` has been set, this may be called + with the syntax ``set_password(user, password)``, + otherwise it must be called with all three arguments: + ``set_password(user, realm, password)``. + :returns: + * ``True`` if existing user was updated + * ``False`` if user account added. + """ + if password is _UNSET: + # called w/ two args - (user, password), use default realm + realm, password = None, realm + user = self._encode_user(user) + realm = self._encode_realm(realm) + key = (user, realm) + existing = (key in self._records) + hash = htdigest.encrypt(password, user, realm, encoding=self.encoding) + if PY3: + hash = hash.encode(self.encoding) + self._records[key] = hash + self._autosave() + return existing + + @deprecated_method(deprecated="1.6", removed="1.8", + replacement="set_password") def update(self, user, realm, password): - """update password for user under specified realm; adding user if needed + "set password for user" + return self.set_password(user, realm, password) + + # XXX: rename to something more explicit, like get_hash()? + def get_hash(self, user, realm=None): + """Return :class:`~passlib.hash.htdigest` hash stored for user. + + * uses ``self.default_realm`` if no realm explicitly provided. + * returns ``None`` if user or realm not found. - :returns: ``True`` if existing user was updated, ``False`` if user added. + .. versionchanged:: 1.6 + This method was previously named ``find``, it was renamed + for clarity. The old name is deprecated, and will be removed + in Passlib 1.8. """ - user = self._norm_user(user) - realm = self._norm_realm(realm) - key = (user,realm) - hash = self._calc_digest(user, realm, password) - return self._update_key(key, hash) + key = (self._encode_user(user), self._encode_realm(realm)) + hash = self._records.get(key) + if hash is None: + return None + if PY3: + hash = hash.decode(self.encoding) + return hash + + @deprecated_method(deprecated="1.6", removed="1.8", + replacement="get_hash") + def find(self, user, realm): + "return hash for user" + return self.get_hash(user, realm) - def delete(self, user, realm): - """delete user's entry for specified realm. + # XXX: rename to something more explicit, like delete_user()? + def delete(self, user, realm=None): + """Delete user's entry for specified realm. - :returns: ``True`` if user deleted, ``False`` if user not found in realm. + if realm is not specified, uses ``self.default_realm``. + + :returns: + * ``True`` if user deleted, + * ``False`` if user not found in realm. """ - user = self._norm_user(user) - realm = self._norm_realm(realm) - return self._delete_key((user,realm)) + key = (self._encode_user(user), self._encode_realm(realm)) + try: + del self._records[key] + except KeyError: + return False + self._autosave() + return True def delete_realm(self, realm): - """delete all users for specified realm + """Delete all users for specified realm. - :returns: number of users deleted + if realm is not specified, uses ``self.default_realm``. + + :returns: number of users deleted (0 if realm not found) """ - realm = self._norm_realm(realm) - keys = [ - key for key in self._entry_map - if key[1] == realm - ] + realm = self._encode_realm(realm) + records = self._records + keys = [key for key in records if key[1] == realm] for key in keys: - self._delete_key(key) + del records[key] + self._autosave() return len(keys) - def find(self, user, realm): - """return digest hash for specified user+realm; returns ``None`` if not found + def check_password(self, user, realm=None, password=_UNSET): + """Verify password for specified user + realm. - :returns: htdigest hash or None - :rtype: bytes or None - """ - user = self._norm_user(user) - realm = self._norm_realm(realm) - hash = self._entry_map.get((user,realm)) - if hash is not None and self.encoding: - #decode hash if in unicode mode - hash = hash.decode("ascii") - return hash - - def verify(self, user, realm, password): - """verify password for specified user + realm. + If ``self.default_realm`` has been set, this may be called + with the syntax ``check_password(user, password)``, + otherwise it must be called with all three arguments: + ``check_password(user, realm, password)``. :returns: - * ``None`` if user not found - * ``False`` if password does not match - * ``True`` if password matches. + * ``None`` if user or realm not found. + * ``False`` if user found, but password does not match. + * ``True`` if user found and password matches. + + .. versionchanged:: 1.6 + This method was previously called ``verify``, it was renamed + to prevent ambiguity with the :class:`!CryptContext` method. + The old alias is deprecated, and will be removed in Passlib 1.8. """ - user = self._norm_user(user) - realm = self._norm_realm(realm) - hash = self._entry_map.get((user,realm)) + if password is _UNSET: + # called w/ two args - (user, password), use default realm + realm, password = None, realm + user = self._encode_user(user) + realm = self._encode_realm(realm) + hash = self._records.get((user,realm)) if hash is None: return None - result = self._calc_digest(user, realm, password) - return consteq(result, hash) + return htdigest.verify(password, hash, user, realm, + encoding=self.encoding) + + @deprecated_method(deprecated="1.6", removed="1.8", + replacement="check_password") + def verify(self, user, realm, password): + "verify password for user" + return self.check_password(user, realm, password) + + #========================================================= + # eoc + #========================================================= #========================================================= # eof diff --git a/passlib/handlers/digests.py b/passlib/handlers/digests.py index ec08056..22c1c6a 100644 --- a/passlib/handlers/digests.py +++ b/passlib/handlers/digests.py @@ -9,7 +9,7 @@ import logging; log = logging.getLogger(__name__) from warnings import warn #site #libs -from passlib.utils import to_native_str +from passlib.utils import to_native_str, to_bytes, render_bytes, consteq from passlib.utils.compat import bascii_to_str, bytes, unicode, str_to_uascii import passlib.utils.handlers as uh from passlib.utils.md4 import md4 @@ -76,5 +76,65 @@ hex_sha256 = create_hex_hash(hashlib.sha256, "sha256") hex_sha512 = create_hex_hash(hashlib.sha512, "sha512") #========================================================= +# htdigest +#========================================================= +class htdigest(object): + """htdigest hash function. + + .. todo:: + document this hash + """ + name = "htdigest" + setting_kwds = () + context_kwds = ("user", "realm") + + @classmethod + def encrypt(cls, secret, user, realm, encoding="utf-8"): + # NOTE: deliberately written so that raw bytes are passed through + # unchanged, encoding only used to handle unicode values. + uh.validate_secret(secret) + if isinstance(secret, unicode): + secret = secret.encode(encoding) + user = to_bytes(user, encoding, "user") + realm = to_bytes(realm, encoding, "realm") + data = render_bytes("%s:%s:%s", user, realm, secret) + return hashlib.md5(data).hexdigest() + + @classmethod + def _norm_hash(cls, hash): + "normalize hash to native string, and validate it" + hash = to_native_str(hash, errname="hash") + if len(hash) != 32: + raise uh.exc.MalformedHashError(cls, "wrong size") + for char in hash: + if char not in uh.LC_HEX_CHARS: + raise uh.exc.MalformedHashError(cls, "invalid chars in hash") + return hash + + @classmethod + def verify(cls, secret, hash, user, realm, encoding="utf-8"): + hash = cls._norm_hash(hash) + other = cls.encrypt(secret, user, realm, encoding) + return consteq(hash, other) + + @classmethod + def identify(cls, hash): + try: + cls._norm_hash(hash) + except ValueError: + return False + return True + + @classmethod + def genconfig(cls): + return None + + @classmethod + def genhash(cls, secret, config, user, realm, encoding="utf-8"): + if config is not None: + cls._norm_hash(config) + return cls.encrypt(secret, user, realm, encoding) + +#========================================================= #eof #========================================================= diff --git a/passlib/registry.py b/passlib/registry.py index 662c9d1..908c23b 100644 --- a/passlib/registry.py +++ b/passlib/registry.py @@ -105,6 +105,7 @@ _handler_locations = { "hex_sha1": ("passlib.handlers.digests", "hex_sha1"), "hex_sha256": ("passlib.handlers.digests", "hex_sha256"), "hex_sha512": ("passlib.handlers.digests", "hex_sha512"), + "htdigest": ("passlib.handlers.digests", "htdigest"), "ldap_plaintext": ("passlib.handlers.ldap_digests","ldap_plaintext"), "ldap_md5": ("passlib.handlers.ldap_digests","ldap_md5"), "ldap_sha1": ("passlib.handlers.ldap_digests","ldap_sha1"), diff --git a/passlib/tests/test_apache.py b/passlib/tests/test_apache.py index d3b4ab8..506ad67 100644 --- a/passlib/tests/test_apache.py +++ b/passlib/tests/test_apache.py @@ -32,10 +32,23 @@ class HtpasswdFileTest(TestCase): "test HtpasswdFile class" descriptionPrefix = "HtpasswdFile" - sample_01 = b('user2:2CHkkwa2AtqGs\nuser3:{SHA}3ipNV1GrBtxPmHFC21fCbVCSXIo=\nuser4:pass4\nuser1:$apr1$t4tc7jTh$GPIWVUo8sQKJlUdV8V5vu0\n') + # sample with 4 users + sample_01 = b('user2:2CHkkwa2AtqGs\n' + 'user3:{SHA}3ipNV1GrBtxPmHFC21fCbVCSXIo=\n' + 'user4:pass4\n' + 'user1:$apr1$t4tc7jTh$GPIWVUo8sQKJlUdV8V5vu0\n') + + # sample 1 with user 1, 2 deleted; 4 changed sample_02 = b('user3:{SHA}3ipNV1GrBtxPmHFC21fCbVCSXIo=\nuser4:pass4\n') - sample_03 = b('user2:pass2x\nuser3:{SHA}3ipNV1GrBtxPmHFC21fCbVCSXIo=\nuser4:pass4\nuser1:$apr1$t4tc7jTh$GPIWVUo8sQKJlUdV8V5vu0\nuser5:pass5\n') + # sample 1 with user2 updated, user 1 first entry removed, and user 5 added + sample_03 = b('user2:pass2x\n' + 'user3:{SHA}3ipNV1GrBtxPmHFC21fCbVCSXIo=\n' + 'user4:pass4\n' + 'user1:$apr1$t4tc7jTh$GPIWVUo8sQKJlUdV8V5vu0\n' + 'user5:pass5\n') + + # standalone sample with 8-bit username sample_04_utf8 = b('user\xc3\xa6:2CHkkwa2AtqGs\n') sample_04_latin1 = b('user\xe6:2CHkkwa2AtqGs\n') @@ -46,60 +59,93 @@ class HtpasswdFileTest(TestCase): if gae_env: return self.skipTest("GAE doesn't offer read/write filesystem access") - #check with existing file + # check with existing file path = mktemp() set_file(path, self.sample_01) ht = apache.HtpasswdFile(path) self.assertEqual(ht.to_string(), self.sample_01) - #check autoload=False - ht = apache.HtpasswdFile(path, autoload=False) + # check without autoload + ht = apache.HtpasswdFile(path, new=True) self.assertEqual(ht.to_string(), b("")) - #check missing file + # check missing file os.remove(path) self.assertRaises(IOError, apache.HtpasswdFile, path) - #NOTE: "default" option checked via update() test, among others + #NOTE: "default_scheme" option checked via set_password() test, among others def test_01_delete(self): "test delete()" - ht = apache.HtpasswdFile._from_string(self.sample_01) - self.assertTrue(ht.delete("user1")) + ht = apache.HtpasswdFile.from_string(self.sample_01) + self.assertTrue(ht.delete("user1")) # should delete both entries self.assertTrue(ht.delete("user2")) - self.assertTrue(not ht.delete("user5")) + self.assertFalse(ht.delete("user5")) # user not present self.assertEqual(ht.to_string(), self.sample_02) + # invalid user self.assertRaises(ValueError, ht.delete, "user:") - def test_02_update(self): - "test update()" - ht = apache.HtpasswdFile._from_string( - self.sample_01, default="plaintext") - self.assertTrue(ht.update("user2", "pass2x")) - self.assertTrue(not ht.update("user5", "pass5")) + def test_01_delete_autosave(self): + if gae_env: + return self.skipTest("GAE doesn't offer read/write filesystem access") + path = mktemp() + sample = b('user1:pass1\nuser2:pass2\n') + set_file(path, sample) + + ht = apache.HtpasswdFile(path) + ht.delete("user1") + self.assertEqual(get_file(path), sample) + + ht = apache.HtpasswdFile(path, autosave=True) + ht.delete("user1") + self.assertEqual(get_file(path), "user2:pass2\n") + + def test_02_set_password(self): + "test set_password()" + ht = apache.HtpasswdFile.from_string( + self.sample_01, default_scheme="plaintext") + self.assertTrue(ht.set_password("user2", "pass2x")) + self.assertFalse(ht.set_password("user5", "pass5")) self.assertEqual(ht.to_string(), self.sample_03) - self.assertRaises(ValueError, ht.update, "user:", "pass") + # invalid user + self.assertRaises(ValueError, ht.set_password, "user:", "pass") + + def test_02_set_password_autosave(self): + if gae_env: + return self.skipTest("GAE doesn't offer read/write filesystem access") + path = mktemp() + sample = b('user1:pass1\n') + set_file(path, sample) + + ht = apache.HtpasswdFile(path) + ht.set_password("user1", "pass2") + self.assertEqual(get_file(path), sample) + + ht = apache.HtpasswdFile(path, default_scheme="plaintext", autosave=True) + ht.set_password("user1", "pass2") + self.assertEqual(get_file(path), "user1:pass2\n") def test_03_users(self): "test users()" - ht = apache.HtpasswdFile._from_string(self.sample_01) - ht.update("user5", "pass5") + ht = apache.HtpasswdFile.from_string(self.sample_01) + ht.set_password("user5", "pass5") ht.delete("user3") - ht.update("user3", "pass3") - self.assertEqual(ht.users(), ["user2", "user4", "user1", "user5", "user3"]) - - def test_04_verify(self): - "test verify()" - ht = apache.HtpasswdFile._from_string(self.sample_01) - self.assertTrue(ht.verify("user5","pass5") is None) + ht.set_password("user3", "pass3") + self.assertEqual(ht.users(), ["user2", "user4", "user1", "user5", + "user3"]) + + def test_04_check_password(self): + "test check_password()" + ht = apache.HtpasswdFile.from_string(self.sample_01) + self.assertTrue(ht.check_password("user5","pass5") is None) for i in irange(1,5): i = str(i) - self.assertTrue(ht.verify("user"+i, "pass"+i)) - self.assertTrue(ht.verify("user"+i, "pass5") is False) + self.assertTrue(ht.check_password("user"+i, "pass"+i)) + self.assertTrue(ht.check_password("user"+i, "pass5") is False) - self.assertRaises(ValueError, ht.verify, "user:", "pass") + self.assertRaises(ValueError, ht.check_password, "user:", "pass") def test_05_load(self): "test load()" @@ -110,33 +156,36 @@ class HtpasswdFileTest(TestCase): path = mktemp() set_file(path, "") backdate_file_mtime(path, 5) - ha = apache.HtpasswdFile(path, default="plaintext") + ha = apache.HtpasswdFile(path, default_scheme="plaintext") self.assertEqual(ha.to_string(), b("")) - #make changes, check force=False does nothing - ha.update("user1", "pass1") - ha.load(force=False) + #make changes, check load_if_changed() does nothing + ha.set_password("user1", "pass1") + ha.load_if_changed() self.assertEqual(ha.to_string(), b("user1:pass1\n")) #change file set_file(path, self.sample_01) - ha.load(force=False) + ha.load_if_changed() self.assertEqual(ha.to_string(), self.sample_01) - #make changes, check force=True overwrites them - ha.update("user5", "pass5") + #make changes, check load() overwrites them + ha.set_password("user5", "pass5") ha.load() self.assertEqual(ha.to_string(), self.sample_01) #test load w/ no path hb = apache.HtpasswdFile() self.assertRaises(RuntimeError, hb.load) - self.assertRaises(RuntimeError, hb.load, force=False) + self.assertRaises(RuntimeError, hb.load_if_changed) - #test load w/ dups + #test load w/ dups and explicit path set_file(path, self.sample_dup) - hc = apache.HtpasswdFile(path) - self.assertTrue(hc.verify('user1','pass1')) + hc = apache.HtpasswdFile() + hc.load(path) + self.assertTrue(hc.check_password('user1','pass1')) + + # NOTE: load_string() tested via from_string(), which is used all over this file def test_06_save(self): "test save()" @@ -155,44 +204,37 @@ class HtpasswdFileTest(TestCase): self.assertEqual(get_file(path), self.sample_02) #test save w/ no path - hb = apache.HtpasswdFile() - hb.update("user1", "pass1") + hb = apache.HtpasswdFile(default_scheme="plaintext") + hb.set_password("user1", "pass1") self.assertRaises(RuntimeError, hb.save) + # test save w/ explicit path + hb.save(path) + self.assertEqual(get_file(path), b("user1:pass1\n")) + def test_07_encodings(self): - "test encoding parameter behavior" - #test bad encodings cause failure in constructor + "test 'encoding' kwd" + # test bad encodings cause failure in constructor self.assertRaises(ValueError, apache.HtpasswdFile, encoding="utf-16") - #check users() returns native string by default - ht = apache.HtpasswdFile._from_string(self.sample_01) - self.assertIsInstance(ht.users()[0], str) - - #check returns unicode if encoding explicitly set - ht = apache.HtpasswdFile._from_string(self.sample_01, encoding="utf-8") - self.assertIsInstance(ht.users()[0], unicode) - - #check returns bytes if encoding explicitly disabled - ht = apache.HtpasswdFile._from_string(self.sample_01, encoding=None) - self.assertIsInstance(ht.users()[0], bytes) - - #check sample utf-8 - ht = apache.HtpasswdFile._from_string(self.sample_04_utf8, encoding="utf-8") + # check sample utf-8 + ht = apache.HtpasswdFile.from_string(self.sample_04_utf8, encoding="utf-8", + return_unicode=True) self.assertEqual(ht.users(), [ u("user\u00e6") ]) - #check sample latin-1 - ht = apache.HtpasswdFile._from_string(self.sample_04_latin1, - encoding="latin-1") + # check sample latin-1 + ht = apache.HtpasswdFile.from_string(self.sample_04_latin1, + encoding="latin-1", return_unicode=True) self.assertEqual(ht.users(), [ u("user\u00e6") ]) def test_08_to_string(self): "test to_string" - #check with known sample - ht = apache.HtpasswdFile._from_string(self.sample_01) + # check with known sample + ht = apache.HtpasswdFile.from_string(self.sample_01) self.assertEqual(ht.to_string(), self.sample_01) - #test blank + # test blank ht = apache.HtpasswdFile() self.assertEqual(ht.to_string(), b("")) @@ -207,10 +249,24 @@ class HtdigestFileTest(TestCase): "test HtdigestFile class" descriptionPrefix = "HtdigestFile" - sample_01 = b('user2:realm:549d2a5f4659ab39a80dac99e159ab19\nuser3:realm:a500bb8c02f6a9170ae46af10c898744\nuser4:realm:ab7b5d5f28ccc7666315f508c7358519\nuser1:realm:2a6cf53e7d8f8cf39d946dc880b14128\n') - sample_02 = b('user3:realm:a500bb8c02f6a9170ae46af10c898744\nuser4:realm:ab7b5d5f28ccc7666315f508c7358519\n') - sample_03 = b('user2:realm:5ba6d8328943c23c64b50f8b29566059\nuser3:realm:a500bb8c02f6a9170ae46af10c898744\nuser4:realm:ab7b5d5f28ccc7666315f508c7358519\nuser1:realm:2a6cf53e7d8f8cf39d946dc880b14128\nuser5:realm:03c55fdc6bf71552356ad401bdb9af19\n') + # sample with 4 users + sample_01 = b('user2:realm:549d2a5f4659ab39a80dac99e159ab19\n' + 'user3:realm:a500bb8c02f6a9170ae46af10c898744\n' + 'user4:realm:ab7b5d5f28ccc7666315f508c7358519\n' + 'user1:realm:2a6cf53e7d8f8cf39d946dc880b14128\n') + + # sample 1 with user 1, 2 deleted; 4 changed + sample_02 = b('user3:realm:a500bb8c02f6a9170ae46af10c898744\n' + 'user4:realm:ab7b5d5f28ccc7666315f508c7358519\n') + # sample 1 with user2 updated, user 1 first entry removed, and user 5 added + sample_03 = b('user2:realm:5ba6d8328943c23c64b50f8b29566059\n' + 'user3:realm:a500bb8c02f6a9170ae46af10c898744\n' + 'user4:realm:ab7b5d5f28ccc7666315f508c7358519\n' + 'user1:realm:2a6cf53e7d8f8cf39d946dc880b14128\n' + 'user5:realm:03c55fdc6bf71552356ad401bdb9af19\n') + + # standalone sample with 8-bit username & realm sample_04_utf8 = b('user\xc3\xa6:realm\xc3\xa6:549d2a5f4659ab39a80dac99e159ab19\n') sample_04_latin1 = b('user\xe6:realm\xe6:549d2a5f4659ab39a80dac99e159ab19\n') @@ -219,61 +275,101 @@ class HtdigestFileTest(TestCase): if gae_env: return self.skipTest("GAE doesn't offer read/write filesystem access") - #check with existing file + # check with existing file path = mktemp() set_file(path, self.sample_01) ht = apache.HtdigestFile(path) self.assertEqual(ht.to_string(), self.sample_01) - #check autoload=False - ht = apache.HtdigestFile(path, autoload=False) + # check without autoload + ht = apache.HtdigestFile(path, new=True) self.assertEqual(ht.to_string(), b("")) - #check missing file + # check missing file os.remove(path) self.assertRaises(IOError, apache.HtdigestFile, path) + # NOTE: default_realm option checked via other tests. + def test_01_delete(self): "test delete()" - ht = apache.HtdigestFile._from_string(self.sample_01) + ht = apache.HtdigestFile.from_string(self.sample_01) self.assertTrue(ht.delete("user1", "realm")) self.assertTrue(ht.delete("user2", "realm")) - self.assertTrue(not ht.delete("user5", "realm")) + self.assertFalse(ht.delete("user5", "realm")) + self.assertFalse(ht.delete("user3", "realm5")) self.assertEqual(ht.to_string(), self.sample_02) + # invalid user self.assertRaises(ValueError, ht.delete, "user:", "realm") - def test_02_update(self): + # invalid realm + self.assertRaises(ValueError, ht.delete, "user", "realm:") + + def test_01_delete_autosave(self): + if gae_env: + return self.skipTest("GAE doesn't offer read/write filesystem access") + path = mktemp() + set_file(path, self.sample_01) + + ht = apache.HtdigestFile(path) + self.assertTrue(ht.delete("user1", "realm")) + self.assertFalse(ht.delete("user3", "realm5")) + self.assertFalse(ht.delete("user5", "realm")) + self.assertEqual(get_file(path), self.sample_01) + + ht.autosave = True + self.assertTrue(ht.delete("user2", "realm")) + self.assertEqual(get_file(path), self.sample_02) + + def test_02_set_password(self): "test update()" - ht = apache.HtdigestFile._from_string(self.sample_01) - self.assertTrue(ht.update("user2", "realm", "pass2x")) - self.assertTrue(not ht.update("user5", "realm", "pass5")) + ht = apache.HtdigestFile.from_string(self.sample_01) + self.assertTrue(ht.set_password("user2", "realm", "pass2x")) + self.assertFalse(ht.set_password("user5", "realm", "pass5")) self.assertEqual(ht.to_string(), self.sample_03) - self.assertRaises(ValueError, ht.update, "user:", "realm", "pass") - self.assertRaises(ValueError, ht.update, "u"*256, "realm", "pass") + # default realm + self.assertRaises(TypeError, ht.set_password, "user2", "pass3") + ht.default_realm = "realm2" + ht.set_password("user2", "pass3") + ht.check_password("user2", "realm2", "pass3") - self.assertRaises(ValueError, ht.update, "user", "realm:", "pass") - self.assertRaises(ValueError, ht.update, "user", "r"*256, "pass") + # invalid user + self.assertRaises(ValueError, ht.set_password, "user:", "realm", "pass") + self.assertRaises(ValueError, ht.set_password, "u"*256, "realm", "pass") + + # invalid realm + self.assertRaises(ValueError, ht.set_password, "user", "realm:", "pass") + self.assertRaises(ValueError, ht.set_password, "user", "r"*256, "pass") + + # TODO: test set_password autosave def test_03_users(self): "test users()" - ht = apache.HtdigestFile._from_string(self.sample_01) - ht.update("user5", "realm", "pass5") + ht = apache.HtdigestFile.from_string(self.sample_01) + ht.set_password("user5", "realm", "pass5") ht.delete("user3", "realm") - ht.update("user3", "realm", "pass3") + ht.set_password("user3", "realm", "pass3") self.assertEqual(ht.users("realm"), ["user2", "user4", "user1", "user5", "user3"]) - def test_04_verify(self): - "test verify()" - ht = apache.HtdigestFile._from_string(self.sample_01) - self.assertTrue(ht.verify("user5", "realm","pass5") is None) + def test_04_check_password(self): + "test check_password()" + ht = apache.HtdigestFile.from_string(self.sample_01) + self.assertIs(ht.check_password("user5", "realm","pass5"), None) for i in irange(1,5): i = str(i) - self.assertTrue(ht.verify("user"+i, "realm", "pass"+i)) - self.assertTrue(ht.verify("user"+i, "realm", "pass5") is False) + self.assertTrue(ht.check_password("user"+i, "realm", "pass"+i)) + self.assertIs(ht.check_password("user"+i, "realm", "pass5"), False) + + # default realm + self.assertRaises(TypeError, ht.check_password, "user5", "pass5") + ht.default_realm = "realm" + self.assertTrue(ht.check_password("user1", "pass1")) + self.assertIs(ht.check_password("user5", "pass5"), None) - self.assertRaises(ValueError, ht.verify, "user:", "realm", "pass") + # invalid user + self.assertRaises(ValueError, ht.check_password, "user:", "realm", "pass") def test_05_load(self): "test load()" @@ -287,25 +383,30 @@ class HtdigestFileTest(TestCase): ha = apache.HtdigestFile(path) self.assertEqual(ha.to_string(), b("")) - #make changes, check force=False does nothing - ha.update("user1", "realm", "pass1") - ha.load(force=False) + #make changes, check load_if_changed() does nothing + ha.set_password("user1", "realm", "pass1") + ha.load_if_changed() self.assertEqual(ha.to_string(), b('user1:realm:2a6cf53e7d8f8cf39d946dc880b14128\n')) #change file set_file(path, self.sample_01) - ha.load(force=False) + ha.load_if_changed() self.assertEqual(ha.to_string(), self.sample_01) #make changes, check force=True overwrites them - ha.update("user5", "realm", "pass5") + ha.set_password("user5", "realm", "pass5") ha.load() self.assertEqual(ha.to_string(), self.sample_01) #test load w/ no path hb = apache.HtdigestFile() self.assertRaises(RuntimeError, hb.load) - self.assertRaises(RuntimeError, hb.load, force=False) + self.assertRaises(RuntimeError, hb.load_if_changed) + + # test load w/ explicit path + hc = apache.HtdigestFile() + hc.load(path) + self.assertEqual(hc.to_string(), self.sample_01) def test_06_save(self): "test save()" @@ -325,12 +426,16 @@ class HtdigestFileTest(TestCase): #test save w/ no path hb = apache.HtdigestFile() - hb.update("user1", "realm", "pass1") + hb.set_password("user1", "realm", "pass1") self.assertRaises(RuntimeError, hb.save) + # test save w/ explicit path + hb.save(path) + self.assertEqual(get_file(path), hb.to_string()) + def test_07_realms(self): "test realms() & delete_realm()" - ht = apache.HtdigestFile._from_string(self.sample_01) + ht = apache.HtdigestFile.from_string(self.sample_01) self.assertEqual(ht.delete_realm("x"), 0) self.assertEqual(ht.realms(), ['realm']) @@ -339,52 +444,36 @@ class HtdigestFileTest(TestCase): self.assertEqual(ht.realms(), []) self.assertEqual(ht.to_string(), b("")) - def test_08_find(self): - "test find()" - ht = apache.HtdigestFile._from_string(self.sample_01) - self.assertEqual(ht.find("user3", "realm"), "a500bb8c02f6a9170ae46af10c898744") - self.assertEqual(ht.find("user4", "realm"), "ab7b5d5f28ccc7666315f508c7358519") - self.assertEqual(ht.find("user5", "realm"), None) + def test_08_get_hash(self): + "test get_hash()" + ht = apache.HtdigestFile.from_string(self.sample_01) + self.assertEqual(ht.get_hash("user3", "realm"), "a500bb8c02f6a9170ae46af10c898744") + self.assertEqual(ht.get_hash("user4", "realm"), "ab7b5d5f28ccc7666315f508c7358519") + self.assertEqual(ht.get_hash("user5", "realm"), None) def test_09_encodings(self): "test encoding parameter" - #test bad encodings cause failure in constructor + # test bad encodings cause failure in constructor self.assertRaises(ValueError, apache.HtdigestFile, encoding="utf-16") - #check users() returns native string by default - ht = apache.HtdigestFile._from_string(self.sample_01) - self.assertIsInstance(ht.realms()[0], str) - self.assertIsInstance(ht.users("realm")[0], str) - - #check returns unicode if encoding explicitly set - ht = apache.HtdigestFile._from_string(self.sample_01, encoding="utf-8") - self.assertIsInstance(ht.realms()[0], unicode) - self.assertIsInstance(ht.users(u("realm"))[0], unicode) - - #check returns bytes if encoding explicitly disabled - ht = apache.HtdigestFile._from_string(self.sample_01, encoding=None) - self.assertIsInstance(ht.realms()[0], bytes) - self.assertIsInstance(ht.users(b("realm"))[0], bytes) - - #check sample utf-8 - ht = apache.HtdigestFile._from_string(self.sample_04_utf8, encoding="utf-8") + # check sample utf-8 + ht = apache.HtdigestFile.from_string(self.sample_04_utf8, encoding="utf-8", return_unicode=True) self.assertEqual(ht.realms(), [ u("realm\u00e6") ]) self.assertEqual(ht.users(u("realm\u00e6")), [ u("user\u00e6") ]) - #check sample latin-1 - ht = apache.HtdigestFile._from_string(self.sample_04_latin1, encoding="latin-1") + # check sample latin-1 + ht = apache.HtdigestFile.from_string(self.sample_04_latin1, encoding="latin-1", return_unicode=True) self.assertEqual(ht.realms(), [ u("realm\u00e6") ]) self.assertEqual(ht.users(u("realm\u00e6")), [ u("user\u00e6") ]) - def test_10_to_string(self): "test to_string()" - #check sample - ht = apache.HtdigestFile._from_string(self.sample_01) + # check sample + ht = apache.HtdigestFile.from_string(self.sample_01) self.assertEqual(ht.to_string(), self.sample_01) - #check blank + # check blank ht = apache.HtdigestFile() self.assertEqual(ht.to_string(), b("")) diff --git a/passlib/tests/test_handlers.py b/passlib/tests/test_handlers.py index 822bc99..b06d100 100644 --- a/passlib/tests/test_handlers.py +++ b/passlib/tests/test_handlers.py @@ -913,6 +913,37 @@ class hex_sha512_test(HandlerCase): ] #========================================================= +# htdigest hash +#========================================================= +class htdigest_test(UserHandlerMixin, HandlerCase): + handler = hash.htdigest + + known_correct_hashes = [ + # secret, user, realm + + # from RFC 2617 + (("Circle Of Life", "Mufasa", "testrealm@host.com"), + '939e7578ed9e3c518a452acee763bce9'), + + # custom + ((UPASS_TABLE, UPASS_USD, UPASS_WAV), + '4dabed2727d583178777fab468dd1f17'), + ] + + def test_80_user(self): + raise self.skipTest("test case doesn't support 'realm' keyword") + + def _insert_user(self, kwds, secret): + "insert username into kwds" + if isinstance(secret, tuple): + secret, user, realm = secret + else: + user, realm = "user", "realm" + kwds.setdefault("user", user) + kwds.setdefault("realm", realm) + return secret + +#========================================================= #ldap hashes #========================================================= class ldap_md5_test(HandlerCase): diff --git a/passlib/utils/__init__.py b/passlib/utils/__init__.py index 9eb8eab..d2fe9a5 100644 --- a/passlib/utils/__init__.py +++ b/passlib/utils/__init__.py @@ -126,30 +126,46 @@ class classproperty(object): "py3 compatible alias" return self.im_func -def deprecated_function(msg=None, deprecated=None, removed=None, updoc=True): +def deprecated_function(msg=None, deprecated=None, removed=None, updoc=True, + replacement=None, _is_method=False): """decorator to deprecate a function. :arg msg: optional msg, default chosen if omitted :kwd deprecated: release where function was first deprecated :kwd removed: release where function will be removed + :kwd replacement: name/instructions for replacement function. :kwd updoc: add notice to docstring (default ``True``) """ if msg is None: - msg = "the function %(mod)s.%(name)s() is deprecated" + if _is_method: + msg = "the method %(mod)s.%(klass)s.%(name)s() is deprecated" + else: + msg = "the function %(mod)s.%(name)s() is deprecated" if deprecated: msg += " as of Passlib %(deprecated)s" if removed: msg += ", and will be removed in Passlib %(removed)s" + if replacement: + msg += ", use %s instead" % replacement msg += "." def build(func): - final = msg % dict( + kwds = dict( mod=func.__module__, name=func.__name__, deprecated=deprecated, removed=removed, - ) + ) + if _is_method: + state = [None] + else: + state = [msg % kwds] def wrapper(*args, **kwds): - warn(final, DeprecationWarning, stacklevel=2) + text = state[0] + if text is None: + klass = args[0].__class__ + kwds.update(klass=klass.__name__, mod=klass.__module__) + text = state[0] = msg % kwds + warn(text, DeprecationWarning, stacklevel=2) return func(*args, **kwds) update_wrapper(wrapper, func) if updoc and (deprecated or removed) and wrapper.__doc__: @@ -162,6 +178,19 @@ def deprecated_function(msg=None, deprecated=None, removed=None, updoc=True): return wrapper return build +def deprecated_method(msg=None, deprecated=None, removed=None, updoc=True, + replacement=None): + """decorator to deprecate a method. + + :arg msg: optional msg, default chosen if omitted + :kwd deprecated: release where function was first deprecated + :kwd removed: release where function will be removed + :kwd replacement: name/instructions for replacement method. + :kwd updoc: add notice to docstring (default ``True``) + """ + return deprecated_function(msg, deprecated, removed, updoc, replacement, + _is_method=True) + ##def relocated_function(target, msg=None, name=None, deprecated=None, mod=None, ## removed=None, updoc=True): ## """constructor to create alias for relocated function. |