""" This software enables you to encode identity data into the findmeon® node standard, to enable cross-network profile tracking and identity tracking. usage: from findmeon_node import SignedInfo from findmeon_node import SeeAlso from findmeon_node import RsaKeyHolder from findmeon_node import generate_node _signedInfo = SignedInfo(resource='https://example.com', resource_type='url', subtype='business', attributes='primary', ) _see_alsos = [SeeAlso(href='https://example.com/foaf.rdf', rel='foaf me'), SeeAlso(href='https://example.com/info.html', rel='xfn me'), ] _rsa_key_private = '''-----BEGIN RSA PRIVATE KEY----- Proc-Type: 4,ENCRYPTED DEK-Info: DES-EDE3-CBC,974B87982C450322 d2IYlyCMdJDDGu4C9WHC1wTbdaqogGWcdpmZk17og3j2e7tQ4JTX0nhMTHHPKvPx 44fQ2VfVxDNqcbJL9xMyGLzCjIz4w/PT3lTNTNRRWMHPBKv4oZ9RHIkTGzrX3l+G KShmXkg1rAhFpiz4eNP7JB/kcZDnSSRE4o+Nvb8w5qirbu8PuJK+kr5u18rC+0OK i+ylsFyBGIGi4poF0Qw1RExSwfPGcBgTaT9jRIJbf/mtaAf2vu6u94G2lGAw5aUv hOrUl2Zjo2l+vACGVF7SW+d/dY85+R2BOZhzuYOmlQm/r9MtUYnxn96oesqwrfu9 YKzGzsycqV+B98srU4dJbjKd/7+z5uJnmJtC1fCC8OFJMaZpWKRuImb5vgyOYAMI BrMfGvi6PjpEE8oHyiiF3KKiaP+HHg+EIaPirNginsHrh3QcdJkbZpefn3NbbfyS 9bsI1P69yH2MLEU/KYSXy9XhmjbwtKUYpyQJOHOmO6J74J7D3rGQl/omG+xSSIX0 r2y2S3Cph/mCv9zVh4ZaishU0VQE/feQNkZzZj/Mr/ck0mqm4kGvP0DJcl8o9XTC aD1YsUGNmGbQMOt330HXmDFfSo8aH3BpcKU40mBw636HIh8gsNDHguEQxEQDx1La cxpcKi/x4bktCW7JBPC/r9aZOy7wNr9vUvKBK8y3WbcDECNbm/puqfAUM5ljOlIA kZSdMQIc9jwAuyrwR4TvcSWHmzIN4P1l6R2KL31ViQxwokrdFpL46eUovIiG69sG qLMvdCqApHakhoed8JcllCws7ulDomv0L88KWCCtrvQQSb4l+PgNyQ== -----END RSA PRIVATE KEY----- ''' _rsa_key_private_passphrase = 'tweet' _rsaKey = RsaKeyHolder(key_private=_rsa_key_private, key_private_passphrase=_rsa_key_private_passphrase, ) encoded_long = generate_node(_signedInfo, _rsaKey, see_alsos=_see_alsos) encoded_long = generate_node(_signedInfo, _rsaKey, see_alsos=_see_alsos, node_type='long') encoded_short = generate_node(_signedInfo, _rsaKey, see_alsos=_see_alsos, node_type='short') encoded_compressed = generate_node(_signedInfo, _rsaKey, see_alsos=_see_alsos, node_type='compressed') Licensing: The MIT License - https://opensource.org/licenses/MIT Copyright 2011-2019 FindMeOn, Inc findmeon®, OpenSN™ and Open Social Network™ are trademarks of FindMeOn, Inc. """ import base64 import datetime import types import urllib from Crypto.PublicKey import RSA from Crypto.Cipher import PKCS1_OAEP SPEC = "https://findmeon.org/spec/0_10" class RsaKeyHolder(object): """wraps an RSA key""" key = None _key_private = None _key_private_passphrase = None key_length_bytes = None block_bytes = None cipher = None def __init__(self, key_private=None, key_private_passphrase=None): self._key_private = key_private self._key_private_passphrase = key_private_passphrase if self._key_private_passphrase: self.key = RSA.importKey(self._key_private, self._key_private_passphrase, ) else: self.key = RSA.importKey(self._key_private) self.key_length_bytes = int((self.key.size() + 1) / 8) # from https://bugs.launchpad.net/pycrypto/+bug/328027 self.block_bytes = self.key_length_bytes - 2 * 20 - 2 self.cipher = PKCS1_OAEP.new(self.key) def encrypt(self, s): encrypted_blocks = [] for block in self.split_string(s, self.block_bytes): encrypted_block = self.cipher.encrypt(block) encrypted_blocks.append(encrypted_block) return ''.join(encrypted_blocks) def decrypt(self, s): decrypted_blocks = [] for block in self.split_string(s, self.key_length_bytes): decrypted_block = self.cipher.decrypt(block) decrypted_blocks.append(decrypted_block) return ''.join(decrypted_blocks) def split_string(self, s, block_size): blocks = [] start = 0 while start < len(s): block = s[start:(start + block_size)] blocks.append(block) start += block_size return blocks class SignedInfo(object): resource = None type = None subtype = None attributes = None _encoded = None # memoize the representation def __init__(self, resource=None, resource_type=None, subtype=None, attributes=None, timestamp=None): self.resource = resource self.resource_type = resource_type self.subtype = subtype self.attributes = attributes self.timestamp = timestamp if timestamp else datetime.datetime.utcnow() @property def encoded(self): if not self._encoded: _assets = [] _assets.append('resource;%s' % self.resource) _assets.append('type;%s' % self.resource_type) if self.subtype: _assets.append('subtype;%s' % self.subtype) if self.attributes: _attributes = self.attributes if isinstance(_attributes, types.StringTypes): _attributes = [_attributes, ] _assets.append('attributes;%s' % ','.join(_attributes)) _assets.append("""timestamp;%s""" % self.timestamp.strftime('%Y-%m-%dT%H:%M:%S') ) self._encoded = '|'.join(_assets) return self._encoded class SeeAlso(object): href = None rel = None def __init__(self, href=None, rel=None): self.href = href self.rel = rel def generate_node(signedInfo, rsaKey, see_alsos=None, node_type='long'): if node_type not in ('long', 'short', 'compressed'): raise ValueError("invalid node_type") _encoded_payload = rsaKey.encrypt(signedInfo.encoded) _padded_b64 = base64.urlsafe_b64encode(_encoded_payload) _padded_b64 = _padded_b64.replace('=', '') # = is a reserved char in urls if node_type in ('long', 'short'): if node_type == 'long': _signed_info = '' _signed_info = _signed_info + ('' % signedInfo.resource) if signedInfo.resource_type: _signed_info = _signed_info + ('' % signedInfo.resource_type) if signedInfo.subtype: _signed_info = _signed_info + ('' % signedInfo.subtype) if signedInfo.attributes: if isinstance(signedInfo.attributes, types.StringTypes): _attributes = [signedInfo.attributes, ] _attributes = ''.join(['''''' % _attr for _attr in signedInfo.attributes]) _signed_info = '''%s''' % _attributes _signed_info = _signed_info + ('''''' % signedInfo.timestamp.strftime('%Y-%m-%dT%H:%M:%S') ) _signed_info = '''%s''' % _signed_info elif node_type == 'short': _signed_info = '''''' % signedInfo.encoded _signature = '' _signature = _signature + '''''' % _padded_b64 _see_also = '' if see_alsos: _see_also = '''%s''' % ''.join(['''''' % (_sa.href, _sa.rel) for _sa in see_alsos]) rendered = '''%(signed_info)s%(signature)s%(see_also)s''' % {'spec': SPEC, 'signed_info': _signed_info, 'signature': _signature, 'see_also': _see_also} return rendered elif node_type == 'compressed': _node = '''''' % { 'spec': urllib.quote_plus(SPEC), 'signed_info': signedInfo.encoded, 'signed_value': _padded_b64, 'see_also': '|'.join(['%s;%s' % (urllib.quote_plus(_sa.rel), urllib.quote_plus(_sa.href)) for _sa in see_alsos]) if see_alsos else '' , } return _node