"""
An example implementation of STROBE. Doesn't include the key tree.
Copyright (c) Mike Hamburg, Cryptography Research, 2016.
I will need to contact legal to get a license for this; in the mean time it is
for example purposes only.
"""
from __future__ import absolute_import
from Strobe.Keccak import KeccakF
class AuthenticationFailed(Exception):
"""Thrown when a MAC fails."""
pass
I,A,C,T,M,K = 1<<0, 1<<1, 1<<2, 1<<3, 1<<4, 1<<5
class Strobe(object):
def __init__(self, proto, F = KeccakF(1600), security = 128, copy_of=None, doInit=True):
if copy_of is None:
self.pos = self.posbegin = 0
self.I0 = None
self.F = F
self.R = F.nbytes - security//4
# Domain separation doesn't use Strobe padding
self.initialized = False
self.st = bytearray(F.nbytes)
domain = bytearray([1,self.R,1,0,1,12*8]) \
+ bytearray(b"STROBEv1.0.2")
if doInit: self._duplex(domain, forceF=True)
# cSHAKE separation is done.
# Turn on Strobe padding and do per-proto separation
self.R -= 2
self.initialized = True
if doInit: self.operate(A|M, proto)
else:
self.R,self.pos,self.posbegin,self.I0,self.F = \
(copy_of.R,copy_of.pos,copy_of.posbegin,copy_of.I0,
copy_of.F.copy())
self.st = bytearray(copy_of.st)
self.initialized = True
def copy(self): return Strobe(None,copy_of=self)
def deepcopy(self): return self.copy()
def set_state_from(self, obj):
self.R,self.pos,self.posbegin,self.I0,self.F = \
(obj.R,obj.pos,obj.posbegin,obj.I0,
obj.F.copy())
self.st = bytearray(obj.st)
self.initialized = True
def _runF(self):
if self.initialized:
self.st[self.pos] ^= self.posbegin
self.st[self.pos+1] ^= 0x04
self.st[self.R+1] ^= 0x80
self.st = self.F(self.st)
self.pos = self.posbegin = 0
def _duplex(self, data, cbefore=False, cafter=False, forceF=False):
assert not (cbefore and cafter)
# Copy data, and convert string or int to bytearray
# This converts an integer n to an array of n zeros
data = bytearray(data)
for i in range(len(data)):
if cbefore: data[i] ^= self.st[self.pos]
self.st[self.pos] ^= data[i]
if cafter: data[i] = self.st[self.pos]
self.pos += 1
if self.pos == self.R: self._runF()
if forceF and self.pos != 0: self._runF()
return data
def _beginOp(self, flags):
# Adjust direction information so that sender and receiver agree
if flags & T:
if self.I0 is None: self.I0 = flags & I
flags ^= self.I0
# Update posbegin
oldbegin, self.posbegin = self.posbegin, self.pos+1
self._duplex([oldbegin,flags], forceF = flags&(C|K))
def operate(self, flags, data, more=False, meta_flags=A|M, metadata=None):
"""
STROBE main duplexing mode.
Op is a byte which describes the operating mode, per the STROBE paper.
Data is either a string or bytearray of data, or else a length. If it
is given as a length, the data is that many bytes of zeros.
If metadata is not None, first apply the given metadata in the given
meta_op.
STROBE operations are streamable. If more is true, this operation
continues the previous operation. It therefore ignores metadata and
doesn't use the beginOp code from the paper.
Certain operations return data. If an operation returns no data
(for example, AD and KEY don't return any data), it returns the empty
byte array.
The meta-operation might also return data. This is convenient for
explicit framing (meta_op = 0b11010/0b11011) or encrypted explicit
framing (meta_op = 0b11110/0b11111)
If the operation is a MAC verification, this function returns the
empty byte array (plus any metadata returned) on success, and throws
AuthenticationFailed on failure.
"""
assert not (flags & (K|1<<6|1<<7)) # Not implemented here
meta_out = bytearray()
if more:
assert flags == self.cur_flags
else:
if metadata is not None:
meta_out = self.operate(meta_flags, metadata)
self._beginOp(flags)
self.cur_flags = flags
if (flags & (I|T) != (I|T)) and (flags & (I|A) != A):
# Operation takes no input
assert isinstance(data,int)
# The actual processing code is just duplex
cafter = (flags & (C|I|T)) == (C|T)
cbefore = (flags & C) and not cafter
processed = self._duplex(data, cbefore, cafter)
# Determine what to do with the output.
if (flags & (I|A)) == (I|A):
# Return data to the application
return meta_out + processed
elif (flags & (I|T)) == T:
# Return data to the transport.
# A fancier implementation might send it directly.
return meta_out + processed
elif (flags & (I|A|T)) == (I|T):
# Check MAC
assert not more
failures = 0
for byte in processed: failures |= byte
if failures != 0: raise AuthenticationFailed()
return meta_out
else:
# Operation has no output data, but maybe output metadata
return meta_out
def ad (self,data, **kw): return self.operate(0b0010,data,**kw)
def key (self,data, **kw): return self.operate(0b0110,data,**kw)
def prf (self,data, **kw): return self.operate(0b0111,data,**kw)
def send_clr(self,data, **kw): return self.operate(0b1010,data,**kw)
def recv_clr(self,data, **kw): return self.operate(0b1011,data,**kw)
def send_enc(self,data, **kw): return self.operate(0b1110,data,**kw)
def recv_enc(self,data, **kw): return self.operate(0b1111,data,**kw)
def send_mac(self,data=16,**kw): return self.operate(0b1100,data,**kw)
def recv_mac(self,data ,**kw): return self.operate(0b1101,data,**kw)
def ratchet (self,data=32,**kw): return self.operate(0b0100,data,**kw)