Source code for spiral.curvecp.address
import collections
from twisted.internet.interfaces import IAddress
from zope.interface import implementer, Attribute
[docs]class ICurveCPAddress(IAddress):
clientExtension = Attribute(
"The 16-byte client extension associated with the connection.")
serverExtension = Attribute(
"The 16-byte server extension associated with the connection.")
serverDomain = Attribute(
"A string representing the server's DNS name.")
longTermKey = Attribute(
"A ``nacl.public.PublicKey`` representing the other side's long term public key.")
transportHost = Attribute(
"The host or IP of the other side of this connection.")
transportPort = Attribute(
"The port of the other side of this connection.")
_CurveCPAddressBase = collections.namedtuple('_CurveCPAddressBase', [
'clientExtension', 'serverExtension', 'serverDomain', 'longTermKey',
'transportHost', 'transportPort',
])
@implementer(ICurveCPAddress)
class CurveCPAddress(_CurveCPAddressBase):
def asUCSPIEnv(self, requesterSide, addressSide):
ret = {
'PROTO': 'CURVECP',
}
if requesterSide == addressSide:
ret.update({
'CURVECPLOCALKEY': str(self.longTermKey).encode('hex'),
'CURVECPLOCALTRANSPORT': '%s:%d' % (
self.transportHost, self.transportPort),
})
else:
ret.update({
'CURVECPREMOTEKEY': str(self.longTermKey).encode('hex'),
'CURVECPREMOTETRANSPORT': '%s:%d' % (
self.transportHost, self.transportPort),
})
if requesterSide == 'client':
ret.update({
'CURVECPREMOTEDOMAIN': self.serverDomain,
'CURVECPLOCALEXTENSION': self.clientExtension.encode('hex'),
'CURVECPREMOTEEXTENSION': self.serverExtension.encode('hex'),
})
elif requesterSide == 'server':
ret.update({
'CURVECPLOCALDOMAIN': self.serverDomain,
'CURVECPREMOTEEXTENSION': self.clientExtension.encode('hex'),
'CURVECPLOCALEXTENSION': self.serverExtension.encode('hex'),
})
return ret