initial commit

README.md

@@ -1,2 +1,2 @@
 # python-scramauth
-Simple implementation of SCRAM authentication
+Simple implementation of SCRAM authentication

scramauth/__init__.py

@@ -0,0 +1,3 @@
+from .session import ScramException, ScramSession
+from .server import ScramServerSession
+from .client import ScramClientSession

scramauth/client.py

@@ -0,0 +1,120 @@
+import base64
+import hashlib
+import hmac
+from collections import OrderedDict
+
+from .session import ScramException, ScramSession
+
+
+class ScramClientSession(ScramSession):
+	def __init__(self, user, password, digest=hashlib.sha256, nonce=None, server_first=None):
+		# init/internal
+		self._step = 0
+		self._digest = digest
+		self._user = user
+		self._password = password
+		self._own_nonce = nonce or self.get_nonce(24)
+		# exchange data
+		self._salt = None
+		self._iterations = None
+		self._nonce = None
+		# jump ahead
+		if server_first is not None:
+			self.create_client_first()
+			self.create_client_final(server_first)
+
+	def process_server_message(self, message):
+		if self._step == 0:
+			self.create_client_first()
+		if self._step == 1:
+			return self.create_client_final(message)
+		elif self._step == 2:
+			return self.check_server(message)
+		else:
+			raise ScramException("Already set up, nothing to do")
+
+	@property
+	def gs2_header(self):
+		return ",".join(["n", ""]) + ","
+
+	@property
+	def client_first_bare(self):
+		return ",".join(["n={}".format(self._user), "r={}".format(self._own_nonce)])
+
+	@property
+	def salted_password(self):
+		if not hasattr(self, "_salted_password"):
+			self._salted_password = self.pbkdf2(self._password, self._salt, self._iterations, digest=self._digest)
+		return self._salted_password
+
+	@property
+	def client_key(self):
+		if not hasattr(self, "_client_key"):
+			self._client_key = hmac.new(self.salted_password, b"Client Key", self._digest).digest()
+		return self._client_key
+
+	@property
+	def stored_key(self):
+		if not hasattr(self, "_stored_key"):
+			self._stored_key = self._digest(self.client_key).digest()
+		return self._stored_key
+
+	@property
+	def server_key(self):
+		if not hasattr(self, "_server_key"):
+			self._server_key = hmac.new(self.salted_password, b"Server Key", self._digest).digest()
+		return self._server_key
+
+	def create_client_first(self):
+		"""
+		Create a client_first message
+
+		:return: str
+		"""
+		self._step = 1
+		self._client_first = self.gs2_header + self.client_first_bare
+		return self._client_first
+
+	def create_client_final(self, server_first):
+		"""
+		Create client_final message based on server_first message.
+
+		:param server_first: str
+		:return: str
+		"""
+		self._server_first = server_first
+		parts = server_first.split(",")
+		received = OrderedDict([tuple(p.split("=", 1)) for i, p in enumerate(parts)])
+		if not received["r"].startswith(self._own_nonce):
+			raise ScramException("Wrong incoming data, nonce does not start with own_nonce: {}.startswith({})", received["r"], self._own_nonce)
+		else:
+			self._nonce = received["r"]
+			self._salt = base64.b64decode(received["s"])
+			self._iterations = int(received["i"])
+			self._client_final_without_proof = ",".join([
+				"c={}".format(base64.b64encode(self.gs2_header.encode()).decode()),
+				"r={}".format(self._nonce)
+			])
+			client_proof = bytes([a ^ b for (a, b) in zip(self.client_key, self.client_signature)])
+			self._client_final = ",".join([self._client_final_without_proof, "p={}".format(base64.b64encode(client_proof).decode())])
+			self._step = 2
+			return self._client_final
+
+	def check_server(self, server_final):
+		"""
+		Check server validation.
+
+		:param server_final: str
+		:return: bool
+		"""
+		self._server_final = server_final
+		parts = server_final.split(",")
+		received = OrderedDict([tuple(p.split("=", 1)) for i, p in enumerate(parts)])
+		if "v" not in received:
+			raise ScramException("Wrong incoming data, validation value not present")
+		else:
+			if received["v"] == base64.b64encode(self.server_signature).decode("utf-8"):
+				self._step = 3
+				return True
+		return False
+

scramauth/server.py

@@ -0,0 +1,131 @@
+import base64
+import hashlib
+from collections import OrderedDict
+
+from .session import ScramException, ScramSession
+
+
+class ScramServerSession(ScramSession):
+	def __init__(self, storage_string, digest=hashlib.sha256, nonce=None, client_first=None):
+		parts = storage_string.split("$")
+		# init/internal
+		self._step = 0
+		self._digest = digest
+		self._salt = base64.b64decode(parts.pop(0))
+		self._iterations = int(parts.pop(0))
+		self._own_nonce = nonce or self.get_nonce(24)
+		# from storage
+		parsed_digests = self.process_storage_string(parts)
+		self._stored_key, self._server_key = parsed_digests.get(digest().name, (None, None))
+		if not self._stored_key or not self._server_key:
+			raise ScramException("Did not find data for digest {}, only {} available".format(digest().name, ", ".join(sorted(parsed_digests.keys()))))
+		# exchange data
+		self._gs2_header = None
+		self._nonce = None
+		# jump ahead
+		if client_first is not None:
+			self.create_server_first(client_first)
+
+	@property
+	def gs2_header(self):
+		return self._gs2_header
+
+	@property
+	def client_key(self):
+		return self._client_key
+
+	@property
+	def stored_key(self):
+		return self._stored_key
+
+	@property
+	def server_key(self):
+		return self._server_key
+
+	def get_client_key(self, client_proof):
+		return bytes(a ^ b for a, b in zip(client_proof, self.client_signature))
+
+	@classmethod
+	def process_storage_string(cls, digest_list):
+		"""
+		Parse parts made by get_storage_keys
+
+		:param digest_list: iterable of get_storage_keys results 
+		:return: dict(digest name, data tuple)
+		"""
+		r = {}
+		for p in digest_list:
+			digest_name, stored_key, server_key = p.split(":")
+			r[digest_name] = (
+				base64.b64decode(stored_key),
+				base64.b64decode(server_key)
+			)
+		return r
+
+	def process_client_message(self, message):
+		"""
+		Generic method to process incoming message, decides what to do based on internal data.
+
+		:param message: str
+		:return: str
+		"""
+		if self._step == 0:
+			return self.create_server_first(message)
+		elif self._step == 1:
+			return self.create_server_final(message)
+		else:
+			raise ScramException("Already set up, nothing to do")
+
+	def create_server_first(self, client_first):
+		"""
+		Prepare server_first message based on client_first.
+
+		:param client_first: str
+		:return: str
+		"""
+		if client_first[0] not in {"n", "p", "y"}:
+			raise ScramException("Wrong client_first, unknown gs2 start")
+		parts = client_first.split(",")
+		data = OrderedDict()
+		if parts[0] == "n" and parts[1] == "":
+			self._client_first = client_first
+			self._gs2_header = ",".join(parts[0:2]) + ","
+			received = OrderedDict([tuple(p.split("=", 1)) for i, p in enumerate(parts) if i > 1])
+			self._nonce = received["r"] + self._own_nonce
+			data["r"] = self._nonce
+			data["s"] = base64.b64encode(self._salt).decode("utf-8")
+			data["i"] = self._iterations
+			self._server_first = ",".join(["{0}={1}".format(*t) for t in data.items()])
+			self._step = 1
+			return self._server_first
+		else:
+			raise ScramException("Unsupported gs2 mode")
+
+	def create_server_final(self, client_final):
+		"""
+		Prepare the server_final message based on client_final.
+
+		If client does not verify properly, raises SCRAMException
+
+		:param client_final: str
+		:return: str
+		"""
+		self._client_final = client_final
+		parts = client_final.split(",")
+		received = OrderedDict([tuple(p.split("=", 1)) for i, p in enumerate(parts)])
+		encoded_gs2_header = base64.b64encode(self.gs2_header.encode("utf-8")).decode("utf-8")
+		if received["c"] != encoded_gs2_header:
+			raise ScramException("Wrong incoming data: {} != {}", received["c"], encoded_gs2_header)
+		elif received["r"] != self._nonce:
+			raise ScramException("Wrong incoming data: {} != {}", received["r"], self._nonce)
+		else:
+			client_proof = base64.b64decode(received["p"])
+			self._client_key = self.get_client_key(client_proof)
+			stored_key = self._digest(self._client_key).digest()
+			if stored_key == self.stored_key:
+				self._server_final = "v={}".format(base64.b64encode(self.server_signature).decode("utf-8"))
+				self._step = 2
+				return self._server_final
+			else:
+				raise ScramException("Invalid client proof {} != {}", stored_key, self.stored_key)
+

scramauth/session.py

@@ -0,0 +1,122 @@
+import base64
+import hashlib
+import hmac
+import os
+
+
+class ScramException(Exception):
+	pass
+
+
+class ScramSession():
+	@classmethod
+	def generate_nonce(cls):
+		"""
+		Return a long enough random blurb
+		
+		:return: bytes
+		"""
+		return os.urandom(64)
+
+	@classmethod
+	def get_storage_string(cls, password, salt, iterations, digests=None):
+		"""
+		Return a string you can store in the database based on password, salt and iteration count.
+		
+		You can pass multiple digests. Older RFCs require support of SHA1, but I really wouldn't, so default is sha256 only.
+		
+		Digest data is result of get_storage_keys.
+		
+		String is of format:
+		[b64 encoded salt]$[iteration count]$[digest 1 data]($[digest N data])
+		
+		:param password: str
+		:param salt: bytes
+		:param iterations: int
+		:param digests: iterable
+		:return: str
+		"""
+		if digests is None:
+			digests = [hashlib.sha256]
+		parts = [base64.b64encode(salt).decode("utf-8"), str(iterations)]
+		for h in digests:
+			parts.append(cls.get_storage_keys(password, salt, iterations, h))
+		return "$".join(parts)
+
+	@classmethod
+	def get_storage_keys(cls, password, salt, iterations, digest):
+		"""
+		Return a : separated string with digest name, base64 encoded stored key and base64 encoded server key
+		
+		:param password: str
+		:param salt: bytes
+		:param iterations: int
+		:param digest: hashlib algorithm constructor
+		:return: str
+		"""
+		salted_password = cls.pbkdf2(password, salt, iterations, digest=digest)
+		client_key = hmac.new(salted_password, b"Client Key", digest).digest()  # 4jTEe/bDZpbdbYUrmaqiuiZVVyg=
+		stored_key = digest(client_key).digest()  # 6dlGYMOdZcOPutkcNY8U2g7vK9Y=
+		server_key = hmac.new(salted_password, b"Server Key", digest).digest()  # D+CSWLOshSulAsxiupA+qs2/fTE='
+		return ":".join([digest().name, base64.b64encode(stored_key).decode("utf-8"), base64.b64encode(server_key).decode("utf-8")])
+
+	@classmethod
+	def pbkdf2(cls, password, salt, iterations, dklen=0, digest=None):
+		"""
+		Return a pbkdf2 value based on inputs
+		
+		:param password: str
+		:param salt: str/bytes
+		:param iterations: int
+		:param dklen: int
+		:param digest: hashlib algorithm constructor
+		:return: bytes
+		"""
+		if digest is None:
+			digest = hashlib.sha256
+		if not dklen:
+			dklen = None
+		elif dklen > (2 ** 32 - 1) * 32:
+			raise OverflowError("dklen too big")
+		password = password.encode("utf-8") if isinstance(password, str) else password
+		salt = salt.encode("utf-8") if isinstance(salt, str) else salt
+		return hashlib.pbkdf2_hmac(digest().name, password, salt, iterations, dklen)
+
+	@classmethod
+	def get_nonce(self, l):
+		"""
+		Return a base64 encoded nonce of specific length.
+		
+		:param l: int
+		:return: str
+		"""
+		return base64.b64encode(self.generate_nonce()).decode("utf-8").strip("=")[0:l]
+
+	@property
+	def client_first_bare(self):
+		return ",".join(self._client_first.split(",")[2:])
+
+	@property
+	def client_final_without_proof(self):
+		if hasattr(self, "_client_final_without_proof"):
+			return self._client_final_without_proof
+		return ",".join([p for p in self._client_final.split(",") if not p.startswith("p=")])
+
+	@property
+	def auth_message(self):
+		if not hasattr(self, "_auth_message"):
+			self._auth_message = ",".join([
+				self.client_first_bare,
+				self._server_first,
+				self.client_final_without_proof
+			]).encode("utf-8")
+		return self._auth_message
+
+	@property
+	def client_signature(self):
+		return hmac.new(self.stored_key, self.auth_message, self._digest).digest()
+
+	@property
+	def server_signature(self):
+		return hmac.new(self.server_key, self.auth_message, self._digest).digest()
+

setup.py

@@ -0,0 +1,12 @@
+from setuptools import setup
+
+setup(
+	name='python-scramauth',
+	version='0.1',
+	url='https://github.com/friedcell/python-scramauth',
+	author='friedcell',
+	author_email='[email protected]',
+	license='MIT',
+	packages=['scramauth'],
+	install_requires=[],
+)