mirror of https://gitlab.com/bashrc2/epicyon
192 lines
6.2 KiB
Python
192 lines
6.2 KiB
Python
__filename__ = "devices.py"
|
|
__author__ = "Bob Mottram"
|
|
__license__ = "AGPL3+"
|
|
__version__ = "1.1.0"
|
|
__maintainer__ = "Bob Mottram"
|
|
__email__ = "bob@freedombone.net"
|
|
__status__ = "Production"
|
|
|
|
# REST API overview
|
|
#
|
|
# To support Olm, the following APIs are required:
|
|
#
|
|
# * Uploading keys for a device (current app)
|
|
# POST /api/v1/crypto/keys/upload
|
|
#
|
|
# * Querying available devices of people you want to establish a session with
|
|
# POST /api/v1/crypto/keys/query
|
|
#
|
|
# * Claiming a pre-key (one-time-key) for each device you want to establish
|
|
# a session with
|
|
# POST /api/v1/crypto/keys/claim
|
|
#
|
|
# * Sending encrypted messages directly to specific devices of other people
|
|
# POST /api/v1/crypto/delivery
|
|
#
|
|
# * Collect encrypted messages addressed to the current device
|
|
# GET /api/v1/crypto/encrypted_messages
|
|
#
|
|
# * Clear all encrypted messages addressed to the current device
|
|
# POST /api/v1/crypto/encrypted_messages/clear
|
|
|
|
import os
|
|
from utils import loadJson
|
|
from utils import saveJson
|
|
|
|
|
|
def E2EEremoveDevice(baseDir: str, nickname: str, domain: str,
|
|
deviceId: str) -> bool:
|
|
"""Unregisters a device for e2ee
|
|
"""
|
|
personDir = baseDir + '/accounts/' + nickname + '@' + domain
|
|
deviceFilename = personDir + '/devices/' + deviceId + '.json'
|
|
if os.path.isfile(deviceFilename):
|
|
os.remove(deviceFilename)
|
|
return True
|
|
return False
|
|
|
|
|
|
def E2EEvalidDevice(deviceJson: {}) -> bool:
|
|
"""Returns true if the given json contains valid device keys
|
|
"""
|
|
if not isinstance(deviceJson, dict):
|
|
return False
|
|
if not deviceJson.get('deviceId'):
|
|
return False
|
|
if not isinstance(deviceJson['deviceId'], str):
|
|
return False
|
|
if not deviceJson.get('type'):
|
|
return False
|
|
if not isinstance(deviceJson['type'], str):
|
|
return False
|
|
if not deviceJson.get('name'):
|
|
return False
|
|
if not isinstance(deviceJson['name'], str):
|
|
return False
|
|
if deviceJson['type'] != 'Device':
|
|
return False
|
|
if not deviceJson.get('claim'):
|
|
return False
|
|
if not isinstance(deviceJson['claim'], str):
|
|
return False
|
|
if not deviceJson.get('fingerprintKey'):
|
|
return False
|
|
if not isinstance(deviceJson['fingerprintKey'], dict):
|
|
return False
|
|
if not deviceJson['fingerprintKey'].get('type'):
|
|
return False
|
|
if not isinstance(deviceJson['fingerprintKey']['type'], str):
|
|
return False
|
|
if not deviceJson['fingerprintKey'].get('publicKeyBase64'):
|
|
return False
|
|
if not isinstance(deviceJson['fingerprintKey']['publicKeyBase64'], str):
|
|
return False
|
|
if not deviceJson.get('identityKey'):
|
|
return False
|
|
if not isinstance(deviceJson['identityKey'], dict):
|
|
return False
|
|
if not deviceJson['identityKey'].get('type'):
|
|
return False
|
|
if not isinstance(deviceJson['identityKey']['type'], str):
|
|
return False
|
|
if not deviceJson['identityKey'].get('publicKeyBase64'):
|
|
return False
|
|
if not isinstance(deviceJson['identityKey']['publicKeyBase64'], str):
|
|
return False
|
|
return True
|
|
|
|
|
|
def E2EEaddDevice(baseDir: str, nickname: str, domain: str,
|
|
deviceId: str, name: str, claimUrl: str,
|
|
fingerprintPublicKey: str,
|
|
identityPublicKey: str,
|
|
fingerprintKeyType="Ed25519Key",
|
|
identityKeyType="Curve25519Key") -> bool:
|
|
"""Registers a device for e2ee
|
|
claimUrl could be something like:
|
|
http://localhost:3000/users/admin/claim?id=11119
|
|
"""
|
|
if ' ' in deviceId or '/' in deviceId or \
|
|
'?' in deviceId or '#' in deviceId or \
|
|
'.' in deviceId:
|
|
return False
|
|
personDir = baseDir + '/accounts/' + nickname + '@' + domain
|
|
if not os.path.isdir(personDir):
|
|
return False
|
|
if not os.path.isdir(personDir + '/devices'):
|
|
os.mkdir(personDir + '/devices')
|
|
deviceDict = {
|
|
"deviceId": deviceId,
|
|
"type": "Device",
|
|
"name": name,
|
|
"claim": claimUrl,
|
|
"fingerprintKey": {
|
|
"type": fingerprintKeyType,
|
|
"publicKeyBase64": fingerprintPublicKey
|
|
},
|
|
"identityKey": {
|
|
"type": identityKeyType,
|
|
"publicKeyBase64": identityPublicKey
|
|
}
|
|
}
|
|
deviceFilename = personDir + '/devices/' + deviceId + '.json'
|
|
return saveJson(deviceDict, deviceFilename)
|
|
|
|
|
|
def E2EEdevicesCollection(baseDir: str, nickname: str, domain: str,
|
|
domainFull: str, httpPrefix: str) -> {}:
|
|
"""Returns a list of registered devices
|
|
"""
|
|
personDir = baseDir + '/accounts/' + nickname + '@' + domain
|
|
if not os.path.isdir(personDir):
|
|
return {}
|
|
personId = httpPrefix + '://' + domainFull + '/users/' + nickname
|
|
if not os.path.isdir(personDir + '/devices'):
|
|
os.mkdir(personDir + '/devices')
|
|
deviceList = []
|
|
for subdir, dirs, files in os.walk(personDir + '/devices/'):
|
|
for dev in files:
|
|
if not dev.endswith('.json'):
|
|
continue
|
|
deviceFilename = os.path.join(personDir + '/devices', dev)
|
|
devJson = loadJson(deviceFilename)
|
|
if devJson:
|
|
deviceList.append(devJson)
|
|
|
|
devicesDict = {
|
|
'id': personId + '/collections/devices',
|
|
'type': 'Collection',
|
|
'totalItems': len(deviceList),
|
|
'items': deviceList
|
|
}
|
|
return devicesDict
|
|
|
|
|
|
def E2EEdecryptMessageFromDevice(messageJson: {}) -> str:
|
|
"""Locally decrypts a message on the device.
|
|
This should probably be a link to a local script
|
|
or native app, such that what the user sees isn't
|
|
something which the server could get access to.
|
|
"""
|
|
# TODO
|
|
# {
|
|
# "type": "EncryptedMessage",
|
|
# "messageType": 0,
|
|
# "cipherText": "...",
|
|
# "digest": {
|
|
# "type": "Digest",
|
|
# "digestAlgorithm": "http://www.w3.org/2000/09/xmldsig#hmac-sha256",
|
|
# "digestValue": "5f6ad31acd64995483d75c7..."
|
|
# },
|
|
# "messageFranking": "...",
|
|
# "attributedTo": {
|
|
# "type": "Device",
|
|
# "deviceId": "11119"
|
|
# },
|
|
# "to": {
|
|
# "type": "Device",
|
|
# "deviceId": "11876"
|
|
# }
|
|
# }
|
|
return ''
|