mirror of
https://gitlab.com/flectra-hq/flectra.git
synced 2026-10-26 16:00:05 -05:00
358 lines
14 KiB
Python
358 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Part of Odoo, Flectra. See LICENSE file for full copyright and licensing details.
|
|
|
|
import datetime
|
|
from enum import Enum
|
|
from importlib import util
|
|
import io
|
|
import json
|
|
import logging
|
|
import netifaces
|
|
from OpenSSL import crypto
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import urllib3
|
|
import zipfile
|
|
from threading import Thread
|
|
import time
|
|
|
|
from flectra import _, http
|
|
from flectra.modules.module import get_resource_path
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
#----------------------------------------------------------
|
|
# Helper
|
|
#----------------------------------------------------------
|
|
|
|
|
|
class CertificateStatus(Enum):
|
|
OK = 1
|
|
NEED_REFRESH = 2
|
|
ERROR = 3
|
|
|
|
|
|
class IoTRestart(Thread):
|
|
"""
|
|
Thread to restart flectra server in IoT Box when we must return a answer before
|
|
"""
|
|
def __init__(self, delay):
|
|
Thread.__init__(self)
|
|
self.delay = delay
|
|
|
|
def run(self):
|
|
time.sleep(self.delay)
|
|
subprocess.check_call(["sudo", "service", "flectra", "restart"])
|
|
|
|
def access_point():
|
|
return get_ip() == '10.11.12.1'
|
|
|
|
def add_credential(db_uuid, enterprise_code):
|
|
write_file('flectra-db-uuid.conf', db_uuid)
|
|
write_file('flectra-enterprise-code.conf', enterprise_code)
|
|
|
|
def check_certificate():
|
|
"""
|
|
Check if the current certificate is up to date or not authenticated
|
|
:return CheckCertificateStatus
|
|
"""
|
|
server = get_flectra_server_url()
|
|
if not server:
|
|
return {"status": CertificateStatus.ERROR,
|
|
"error_code": "ERR_IOT_HTTPS_CHECK_NO_SERVER"}
|
|
|
|
path = Path('/etc/ssl/certs/nginx-cert.crt')
|
|
if not path.exists():
|
|
return {"status": CertificateStatus.NEED_REFRESH}
|
|
|
|
try:
|
|
with path.open('r') as f:
|
|
cert = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
|
|
except EnvironmentError:
|
|
_logger.exception("Unable to read certificate file")
|
|
return {"status": CertificateStatus.ERROR,
|
|
"error_code": "ERR_IOT_HTTPS_CHECK_CERT_READ_EXCEPTION"}
|
|
|
|
cert_end_date = datetime.datetime.strptime(cert.get_notAfter().decode('utf-8'), "%Y%m%d%H%M%SZ") - datetime.timedelta(days=10)
|
|
for key in cert.get_subject().get_components():
|
|
if key[0] == b'CN':
|
|
cn = key[1].decode('utf-8')
|
|
if cn == 'FlectraTempIoTBoxCertificate' or datetime.datetime.now() > cert_end_date:
|
|
message = _('Your certificate %s must be updated') % (cn)
|
|
_logger.info(message)
|
|
return {"status": CertificateStatus.NEED_REFRESH}
|
|
else:
|
|
message = _('Your certificate %s is valid until %s') % (cn, cert_end_date)
|
|
_logger.info(message)
|
|
return {"status": CertificateStatus.OK, "message": message}
|
|
|
|
def check_git_branch():
|
|
"""
|
|
Check if the local branch is the same than the connected Flectra DB and
|
|
checkout to match it if needed.
|
|
"""
|
|
server = get_flectra_server_url()
|
|
if server:
|
|
urllib3.disable_warnings()
|
|
http = urllib3.PoolManager(cert_reqs='CERT_NONE')
|
|
try:
|
|
response = http.request(
|
|
'POST',
|
|
server + "/web/webclient/version_info",
|
|
body = '{}',
|
|
headers = {'Content-type': 'application/json'}
|
|
)
|
|
|
|
if response.status == 200:
|
|
git = ['git', '--work-tree=/home/pi/flectra/', '--git-dir=/home/pi/flectra/.git']
|
|
|
|
db_branch = json.loads(response.data)['result']['server_serie'].replace('~', '-')
|
|
if not subprocess.check_output(git + ['ls-remote', 'origin', db_branch]):
|
|
db_branch = 'master'
|
|
|
|
local_branch = subprocess.check_output(git + ['symbolic-ref', '-q', '--short', 'HEAD']).decode('utf-8').rstrip()
|
|
|
|
if db_branch != local_branch:
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/"])
|
|
subprocess.check_call(["rm", "-rf", "/home/pi/flectra/addons/hw_drivers/iot_handlers/drivers/*"])
|
|
subprocess.check_call(["rm", "-rf", "/home/pi/flectra/addons/hw_drivers/iot_handlers/interfaces/*"])
|
|
subprocess.check_call(git + ['branch', '-m', db_branch])
|
|
subprocess.check_call(git + ['remote', 'set-branches', 'origin', db_branch])
|
|
os.system('/home/pi/flectra/addons/point_of_sale/tools/posbox/configuration/posbox_update.sh')
|
|
subprocess.call(["sudo", "mount", "-o", "remount,ro", "/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/root_bypass_ramdisks/etc/cups"])
|
|
|
|
except Exception as e:
|
|
_logger.error('Could not reach configured server')
|
|
_logger.error('A error encountered : %s ' % e)
|
|
|
|
def check_image():
|
|
"""
|
|
Check if the current image of IoT Box is up to date
|
|
"""
|
|
url = 'https://nightly.flectrahq.com/master/iotbox/SHA1SUMS.txt'
|
|
urllib3.disable_warnings()
|
|
http = urllib3.PoolManager(cert_reqs='CERT_NONE')
|
|
response = http.request('GET', url)
|
|
checkFile = {}
|
|
valueActual = ''
|
|
for line in response.data.decode().split('\n'):
|
|
if line:
|
|
value, name = line.split(' ')
|
|
checkFile.update({value: name})
|
|
if name == 'iotbox-latest.zip':
|
|
valueLastest = value
|
|
elif name == get_img_name():
|
|
valueActual = value
|
|
if valueActual == valueLastest:
|
|
return False
|
|
version = checkFile.get(valueLastest, 'Error').replace('iotboxv', '').replace('.zip', '').split('_')
|
|
return {'major': version[0], 'minor': version[1]}
|
|
|
|
def get_certificate_status(is_first=True):
|
|
"""
|
|
Will get the HTTPS certificate details if present. Will load the certificate if missing.
|
|
|
|
:param is_first: Use to make sure that the recursion happens only once
|
|
:return: (bool, str)
|
|
"""
|
|
check_certificate_result = check_certificate()
|
|
certificateStatus = check_certificate_result["status"]
|
|
|
|
if certificateStatus == CertificateStatus.ERROR:
|
|
return False, check_certificate_result["error_code"]
|
|
|
|
if certificateStatus == CertificateStatus.NEED_REFRESH and is_first:
|
|
certificate_process = load_certificate()
|
|
if certificate_process is not True:
|
|
return False, certificate_process
|
|
return get_certificate_status(is_first=False) # recursive call to attempt certificate read
|
|
return True, check_certificate_result.get("message",
|
|
"The HTTPS certificate was generated correctly")
|
|
|
|
def get_img_name():
|
|
major, minor = get_version().split('.')
|
|
return 'iotboxv%s_%s.zip' % (major, minor)
|
|
|
|
def get_ip():
|
|
while True:
|
|
try:
|
|
return netifaces.ifaddresses('eth0')[netifaces.AF_INET][0]['addr']
|
|
except KeyError:
|
|
pass
|
|
|
|
try:
|
|
return netifaces.ifaddresses('wlan0')[netifaces.AF_INET][0]['addr']
|
|
except KeyError:
|
|
pass
|
|
|
|
_logger.warning("Couldn't get IP, sleeping and retrying.")
|
|
time.sleep(5)
|
|
|
|
def get_mac_address():
|
|
while True:
|
|
try:
|
|
return netifaces.ifaddresses('eth0')[netifaces.AF_LINK][0]['addr']
|
|
except KeyError:
|
|
pass
|
|
|
|
try:
|
|
return netifaces.ifaddresses('wlan0')[netifaces.AF_LINK][0]['addr']
|
|
except KeyError:
|
|
pass
|
|
|
|
_logger.warning("Couldn't get MAC address, sleeping and retrying.")
|
|
time.sleep(5)
|
|
|
|
def get_ssid():
|
|
ap = subprocess.call(['systemctl', 'is-active', '--quiet', 'hostapd']) # if service is active return 0 else inactive
|
|
if not ap:
|
|
return subprocess.check_output(['grep', '-oP', '(?<=ssid=).*', '/etc/hostapd/hostapd.conf']).decode('utf-8').rstrip()
|
|
process_iwconfig = subprocess.Popen(['iwconfig'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
|
process_grep = subprocess.Popen(['grep', 'ESSID:"'], stdin=process_iwconfig.stdout, stdout=subprocess.PIPE)
|
|
return subprocess.check_output(['sed', 's/.*"\\(.*\\)"/\\1/'], stdin=process_grep.stdout).decode('utf-8').rstrip()
|
|
|
|
def get_flectra_server_url():
|
|
ap = subprocess.call(['systemctl', 'is-active', '--quiet', 'hostapd']) # if service is active return 0 else inactive
|
|
if not ap:
|
|
return False
|
|
return read_file_first_line('flectra-remote-server.conf')
|
|
|
|
def get_token():
|
|
return read_file_first_line('token')
|
|
|
|
def get_version():
|
|
return subprocess.check_output(['cat', '/var/flectra/iotbox_version']).decode().rstrip()
|
|
|
|
def get_wifi_essid():
|
|
wifi_options = []
|
|
process_iwlist = subprocess.Popen(['sudo', 'iwlist', 'wlan0', 'scan'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
|
process_grep = subprocess.Popen(['grep', 'ESSID:"'], stdin=process_iwlist.stdout, stdout=subprocess.PIPE).stdout.readlines()
|
|
for ssid in process_grep:
|
|
essid = ssid.decode('utf-8').split('"')[1]
|
|
if essid not in wifi_options:
|
|
wifi_options.append(essid)
|
|
return wifi_options
|
|
|
|
def load_certificate():
|
|
"""
|
|
Send a request to Flectra with customer db_uuid and enterprise_code to get a true certificate
|
|
"""
|
|
db_uuid = read_file_first_line('flectra-db-uuid.conf')
|
|
enterprise_code = read_file_first_line('flectra-enterprise-code.conf')
|
|
if not (db_uuid and enterprise_code):
|
|
return "ERR_IOT_HTTPS_LOAD_NO_CREDENTIAL"
|
|
|
|
url = 'https://www.flectrahq.com/flectra-enterprise/iot/x509'
|
|
data = {
|
|
'params': {
|
|
'db_uuid': db_uuid,
|
|
'enterprise_code': enterprise_code
|
|
}
|
|
}
|
|
urllib3.disable_warnings()
|
|
http = urllib3.PoolManager(cert_reqs='CERT_NONE', retries=urllib3.Retry(4))
|
|
try:
|
|
response = http.request(
|
|
'POST',
|
|
url,
|
|
body = json.dumps(data).encode('utf8'),
|
|
headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
|
|
)
|
|
except Exception as e:
|
|
_logger.exception("An error occurred while trying to reach flectrahq.com servers.")
|
|
return "ERR_IOT_HTTPS_LOAD_REQUEST_EXCEPTION\n\n%s" % e
|
|
|
|
if response.status != 200:
|
|
return "ERR_IOT_HTTPS_LOAD_REQUEST_STATUS %s\n\n%s" % (response.status, response.reason)
|
|
|
|
result = json.loads(response.data.decode('utf8'))['result']
|
|
if not result:
|
|
return "ERR_IOT_HTTPS_LOAD_REQUEST_NO_RESULT"
|
|
|
|
write_file('flectra-subject.conf', result['subject_cn'])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/root_bypass_ramdisks/"])
|
|
Path('/etc/ssl/certs/nginx-cert.crt').write_text(result['x509_pem'])
|
|
Path('/root_bypass_ramdisks/etc/ssl/certs/nginx-cert.crt').write_text(result['x509_pem'])
|
|
Path('/etc/ssl/private/nginx-cert.key').write_text(result['private_key_pem'])
|
|
Path('/root_bypass_ramdisks/etc/ssl/private/nginx-cert.key').write_text(result['private_key_pem'])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,ro", "/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,ro", "/root_bypass_ramdisks/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/root_bypass_ramdisks/etc/cups"])
|
|
subprocess.check_call(["sudo", "service", "nginx", "restart"])
|
|
return True
|
|
|
|
def download_iot_handlers(auto=True):
|
|
"""
|
|
Get the drivers from the configured Flectra server
|
|
"""
|
|
server = get_flectra_server_url()
|
|
if server:
|
|
urllib3.disable_warnings()
|
|
pm = urllib3.PoolManager(cert_reqs='CERT_NONE')
|
|
server = server + '/iot/get_handlers'
|
|
try:
|
|
resp = pm.request('POST', server, fields={'mac': get_mac_address(), 'auto': auto})
|
|
if resp.data:
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/"])
|
|
drivers_path = Path.home() / 'flectra/addons/hw_drivers/iot_handlers'
|
|
zip_file = zipfile.ZipFile(io.BytesIO(resp.data))
|
|
zip_file.extractall(drivers_path)
|
|
subprocess.call(["sudo", "mount", "-o", "remount,ro", "/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/root_bypass_ramdisks/etc/cups"])
|
|
except Exception as e:
|
|
_logger.error('Could not reach configured server')
|
|
_logger.error('A error encountered : %s ' % e)
|
|
|
|
def compute_iot_handlers_addon_name(handler_kind, handler_file_name):
|
|
# TODO: replace with `removesuffix` (for Flectra version using an IoT image that use Python >= 3.9)
|
|
return "flectra.addons.hw_drivers.iot_handlers.{handler_kind}.{handler_name}".\
|
|
format(handler_kind=handler_kind, handler_name=handler_file_name.replace('.py', ''))
|
|
|
|
def load_iot_handlers():
|
|
"""
|
|
This method loads local files: 'flectra/addons/hw_drivers/iot_handlers/drivers' and
|
|
'flectra/addons/hw_drivers/iot_handlers/interfaces'
|
|
And execute these python drivers and interfaces
|
|
"""
|
|
for directory in ['interfaces', 'drivers']:
|
|
path = get_resource_path('hw_drivers', 'iot_handlers', directory)
|
|
filesList = os.listdir(path)
|
|
for file in filesList:
|
|
path_file = os.path.join(path, file)
|
|
spec = util.spec_from_file_location(compute_iot_handlers_addon_name(directory, file), path_file)
|
|
if spec:
|
|
module = util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
http.addons_manifest = {}
|
|
http.root = http.Root()
|
|
|
|
def flectra_restart(delay):
|
|
IR = IoTRestart(delay)
|
|
IR.start()
|
|
|
|
def read_file_first_line(filename):
|
|
path = Path.home() / filename
|
|
path = Path('/home/pi/' + filename)
|
|
if path.exists():
|
|
with path.open('r') as f:
|
|
return f.readline().strip('\n')
|
|
return ''
|
|
|
|
def unlink_file(filename):
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/"])
|
|
path = Path.home() / filename
|
|
if path.exists():
|
|
path.unlink()
|
|
subprocess.call(["sudo", "mount", "-o", "remount,ro", "/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/root_bypass_ramdisks/etc/cups"])
|
|
|
|
def write_file(filename, text):
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/"])
|
|
path = Path.home() / filename
|
|
path.write_text(text)
|
|
subprocess.call(["sudo", "mount", "-o", "remount,ro", "/"])
|
|
subprocess.call(["sudo", "mount", "-o", "remount,rw", "/root_bypass_ramdisks/etc/cups"])
|