[Zodb-checkins] CVS: ZODB3/ZEO/auth - auth_digest.py:1.1.2.1

[Jeremy Hylton]

Update of /cvs-repository/ZODB3/ZEO/auth
In directory cvs.zope.org:/tmp/cvs-serv10189/ZEO/auth

Added Files:
      Tag: ZODB3-auth-branch
	auth_digest.py 
Log Message:
Add a simple digest authentication based on HTTP digest authentication.


=== Added File ZODB3/ZEO/auth/auth_digest.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL).  A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE
#
##############################################################################
"""Digest authentication for ZEO

This authentication mechanism follows the design of HTTP digest
authentication (RFC 2069).  It is a simple challenge-response protocol
that does not send passwords in the clear, but does not offer strong
security.  The RFC discusses many of the limitations of this kind of
protocol.

XXX The HTTP mechanism includes a "realm" that specifies in what
context the username is valid.  ZEO does not have any notion of a
realm at the moment, but it probably should.  The password database
should store the hash of user, realm, and password.  Right now, it
just stores the hash of user and password, which would allow an
attacker to determine that the user-password combo was the same at two
different sites.

Guard the password database as if it contained plaintext passwords.
It stores the hash of a username and password.  This does not expose
the plaintext password, but it is sensitive nonetheless.  An attack
with the hash can impersonate the real user.  This is a limitation of
the simple digest scheme.
"""

import base64
import os
import sha
import struct
import time

from ZEO.auth.storage import AuthZEOStorage
from ZEO.auth.database import Database

def get_random_bytes(n):
    if os.path.exists("/dev/urandom"):
        f = open("/dev/urandom")
        s = f.read(8)
        f.close()
    else:
        L = [chr(random.randint(0, 255)) for i in range(8)]
        s = "".join(L)
    return s

def hexdigest(s):
    return sha.new(s).hexdigest()

class DigestDatabase(Database):

    def __init__(self, filename):
        Database.__init__(self, filename)
        
        # Initialize a key used to build the nonce for a challenge.
        # We need one key for the lifetime of the server.
        self.noncekey = get_random_bytes(8)

    def _store_password(self, username, password):
        dig = hexdigest("%s:%s" % (username, password))
        self._users[username] = dig

class StorageClass(AuthZEOStorage):

    def set_database(self, database):
        assert isinstance(database, DigestDatabase)
        self.database = database
        self.noncekey = database.noncekey

    def _get_time(self):
        # Return an approximate time value suitable for a nonce.
        # We check that the nonce is valid by generating a new
        # one and comparing it with a previously generated one.
        # So the nonce time should change infrequently.  If a client
        # receives a nonce on the cusp of a time-change, it will
        # need to request a new authentication challenge.

        # Start with once an hour
        t = int(time.time()) / 3600
        return struct.pack("i", t)

    def _get_nonce(self):
        # RFC 2069 recommends a nonce of the form
        # H(client-IP ":" time-stamp ":" private-key)
        dig = sha.sha()
        dig.update(str(self.connection.addr))
        dig.update(self._get_time())
        dig.update(self.noncekey)
        self._nonce = dig.hexdigest()
        return self._nonce

    def auth_get_challenge(self):
        """Return a nonce as the challenge."""
        # XXX perhaps the server should identify itself in
        # the challenge
        return self._get_nonce()

    def auth_response(self, resp):
        # verify client response
        user, nonce, response = resp

        # Since zrpc is a stateful protocol, we just store the nonce
        # we sent to the client.  It will need to generate a new
        # nonce for a new connection anyway.
        if self._nonce != nonce:
            raise ValueError, "invalid nonce"

        # lookup user in database
        h_up = self.database.get_password(user)

        # regeneration resp from user, password, and nonce
        check = hexdigest("%s:%s" % (h_up, nonce))
        return self.finish_auth(check == response)

class Client:

    def __init__(self, stub):
        self.stub = stub

    def auth_get_challenge(self):
        m = self.stub.extensionMethod("auth_get_challenge")
        return m()

    def auth_response(self, t):
        m = self.stub.extensionMethod("auth_response")
        return m(t)
    
    def start(self, username, password):
        h_up = hexdigest("%s:%s" % (username, password))
        nonce = self.auth_get_challenge()
        
        resp_dig = hexdigest("%s:%s" % (h_up, nonce))
        return self.auth_response((username, nonce, resp_dig))
    
# StorageServer checks here for DatabaseClass
DatabaseClass = DigestDatabase