This commit is contained in:
Arti Zirk 2020-11-03 21:18:35 +02:00
commit d2de4e13d7
9 changed files with 440 additions and 0 deletions

66
.gitignore vendored Normal file
View File

@ -0,0 +1,66 @@
# ---> Python
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*,cover
# Translations
*.mo
*.pot
# Django stuff:
*.log
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# PyCharm
.idea/
# Other
venv/
run.sh

18
README.md Normal file
View File

@ -0,0 +1,18 @@
# K-Door-Pi
Client for K-Space door system. Run on Raspberry Pi and controlls the doors
# Running it
```sh
export KDOORPI_DOOR="name" # Door name
export KDOORPI_API_ALLOWED="http://server/allowed" # Used to fetch allowed cards
export KDOORPI_API_LONGPOLL="http://server/longpoll" # Open door now and force allowed cards sync endpoint
export KDOORPI_API_SWIPE="http://server/swipe" # On card swipe post log line here
export KDOORPI_API_KEY="keykeykey" # api key for accessing allowed and longpoll api endpoints
export KDOORPI_UID_SALT="saltsaltsalt" # shared slat used to hash card uid bytes
python3 -m kdoorpi
```
You can also use the `run.sh.example` by copying it to `run.sh` and changing variables inside it.

26
contrib/uidhash.py Normal file
View File

@ -0,0 +1,26 @@
import secrets
from functools import partial
try:
from hashlib import scrypt
uid_hash = partial(scrypt, n=16384, r=8, p=1)
except ImportError:
# Python 3.5 has to use external py-scrypt package from pypi
from scrypt import hash as scrypt
uid_hash = partial(scrypt, N=16384, r=8, p=1)
# print(secrets.token_urlsafe())
UID_SALT = "hkRXwLlQKmCJoy5qaahp"
def hash_uid(uid: str, salt: str = UID_SALT) -> str:
return uid_hash(bytes.fromhex(uid), salt=salt.encode()).hex()
if __name__ == "__main__":
UID = "01 23 AD F0"
ch = hash_uid(UID)
h = 'a6d9ba36ecb5f8e6312f40ee260ad59e9cca3c6ce073bf072df3324c0072886196e6823a7c758ab567fc53e91fbbda297a4efe0072e41316c56446ef126a5180'
print("UID:", UID)
print("hash:", ch)
print("correct", ch == h)

114
kdoorpi/__init__.py Normal file
View File

@ -0,0 +1,114 @@
import logging
import threading
import requests
import time
from functools import partial
from json.decoder import JSONDecodeError
try:
from hashlib import scrypt
UID_HASH = partial(scrypt, n=16384, r=8, p=1)
except ImportError:
# Python 3.5 needs pip install scrypt
from scrypt import hash as scrypt
UID_HASH = partial(scrypt, N=16384, r=8, p=1)
from .wiegand import Decoder
logging.basicConfig(level=logging.DEBUG)
def hash_uid(uid: str, salt: str) -> str:
return UID_HASH(bytes.fromhex(uid), salt=salt.encode()).hex()
class Main:
def __init__(self,
door,
api_allowed,
api_longpoll,
api_swipe,
api_key,
uid_hash):
self.door = door
self.api_allowed = api_allowed
self.api_longpoll = api_longpoll
self.api_swipe = api_swipe
self.uid_hash = uid_hash
self.uids = {}
self.force_sync_now = threading.Event()
self.session = requests.Session()
self.session.headers.update({"KEY": api_key})
self.sync_cards()
logging.info("Running")
self.wiegand = Decoder(self.wiegand_callback)
self.notify_thread = threading.Thread(target=self.listen_notification, daemon=True)
self.notify_thread.start()
self.auto_sync_loop()
def sync_cards(self):
logging.info("Downloading users list")
r = self.session.get(self.api_allowed)
try:
allowed_uids = r.json()["allowed_uids"]
except JSONDecodeError as e:
logging.exception("Failed to decode allowed uids json")
return
uids = set()
for token in allowed_uids:
uids.add(token["token"]["uid_hash"].strip())
self.uids = uids
def wiegand_callback(self, bits, value):
#print("bits", bits, "value", value)
uid_h = hash_uid(value, self.uid_hash)
logging.debug("hash %s", uid_h)
if uid_h in self.uids:
logging.info("Opening door for UID hash trail %s", uid_h[-10:])
self.wiegand.open_door()
success = True
else:
logging.info("Access card not in allow list, hash trail %s", uid_h[-10:])
success = False
data = {
"uid": value,
"door": self.door,
"success": success,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
}
print(data)
requests.post(self.api_swipe, data=data)
def listen_notification(self):
while 1:
try:
r = self.session.get(self.api_longpoll,
headers={"Connection": "close"},
timeout=60, stream=True)
for line in r.iter_lines(1, decode_unicode=True):
if not line.strip():
continue
logging.debug("Got notification: %s", line)
if self.door in line:
logging.info("Opening door from notify")
self.wiegand.open_door()
self.force_sync_now.set()
except (requests.Timeout,
requests.ConnectionError) as e:
logging.debug("notification timeout")
time.sleep(0.1)
def auto_sync_loop(self):
while 1:
try:
self.force_sync_now.wait(60*10) # == 10min
self.force_sync_now.clear()
self.sync_cards()
except KeyboardInterrupt as e:
self.wiegand.cancel()
break

30
kdoorpi/__main__.py Normal file
View File

@ -0,0 +1,30 @@
import os
from . import Main
"""
if "eesuks" in buf:
door = "front"
elif "valis" in buf:
door = "ground"
elif "taga" in buf:
door = "back"
else:
door = "unknown"
"""
door = os.environ["KDOORPI_DOOR"]
api_allowed = os.environ["KDOORPI_API_ALLOWED"]
api_longpoll = os.environ["KDOORPI_API_LONGPOLL"]
api_swipe = os.environ["KDOORPI_API_SWIPE"]
api_key = os.environ["KDOORPI_API_KEY"]
uid_salt = os.environ["KDOORPI_UID_SALT"]
if __name__ == "__main__":
Main(door,
api_allowed,
api_longpoll,
api_swipe,
api_key,
uid_salt)

150
kdoorpi/wiegand.py Normal file
View File

@ -0,0 +1,150 @@
#!/bin/python
import logging
from time import sleep
try:
import pigpio
except Exception:
pigpio = False
class Decoder:
"""
A class to read Wiegand codes of an arbitrary length.
"""
def __init__(self, callback, bit_timeout=5):
"""
Instantiate with the pi, gpio for 0 (green wire), the gpio for 1
(white wire), the callback function, and the bit timeout in
milliseconds which indicates the end of a code.
The callback is passed the code length in bits and the hex value.
"""
if pigpio:
self.pi = pigpio.pi()
else:
self.pi = False
self.gpio_0 = 17 #settings.WIEGAND[0]
self.gpio_1 = 18 #settings.WIEGAND[1]
self.door_pin = 21 # from settings.py
self.button_pin = 13 # from settings.py
self.callback = callback
self.bit_timeout = bit_timeout
self.items = []
self.in_code = False
if self.pi:
self.pi.set_mode(self.gpio_0, pigpio.INPUT)
self.pi.set_mode(self.gpio_1, pigpio.INPUT)
self.pi.set_mode(self.door_pin, pigpio.OUTPUT)
self.pi.set_mode(self.button_pin, pigpio.INPUT)
self.pi.set_pull_up_down(self.gpio_0, pigpio.PUD_UP)
self.pi.set_pull_up_down(self.gpio_1, pigpio.PUD_UP)
self.pi.set_pull_up_down(self.button_pin, pigpio.PUD_UP)
self.cb_0 = self.pi.callback(self.gpio_0, pigpio.FALLING_EDGE, self._cb)
self.cb_1 = self.pi.callback(self.gpio_1, pigpio.FALLING_EDGE, self._cb)
self.button_cb_h = self.pi.callback(self.button_pin, pigpio.FALLING_EDGE, self._cb)
def cut_empty(self, item):
if item[0:8] == "00000000":
return self.cut_empty(item[8:])
else:
return item
def get_hex(self):
try:
items = self.items
if len(self.items) == 26:
items = self.items[1:-1]
elif len(self.items) == 64:
items = self.cut_empty(self.items)
bits = []
for i in range(len(items), 0, -8):
bits.append(int(items[i - 8:i], 2))
return (" ".join(map(lambda a: "%-0.2X" % ((a + 256) % 256), bits))).rstrip()
except ValueError:
logging.error("Wiegand convert error: bin to hex convertion ended with ValeError. raw: " + str(self.items))
return False
except Exception as e:
logging.error("Wiegand convert error: (raw: " + str(self.items) + ") " + str(e))
return False
def _cb(self, gpio_pin, level, tick):
"""
Accumulate bits until both gpios 0 and 1 timeout.
"""
try:
if level < pigpio.TIMEOUT:
if self.in_code:
self.bits += 1
else:
logging.debug("Wiegand data transfer start")
self.bits = 1
self.items = ""
self.in_code = True
self.code_timeout = 0
self.pi.set_watchdog(self.gpio_0, self.bit_timeout)
self.pi.set_watchdog(self.gpio_1, self.bit_timeout)
if gpio_pin == self.gpio_0:
self.code_timeout &= 2 # clear gpio 0 timeout
self.items += "1"
else:
self.code_timeout &= 1 # clear gpio 1 timeout
self.items += "0"
else:
if self.in_code:
if gpio_pin == self.gpio_0:
self.code_timeout |= 1 # timeout gpio 0
else:
self.code_timeout |= 2 # timeout gpio 1
if self.code_timeout == 3: # both gpios timed out
self.pi.set_watchdog(self.gpio_0, 0)
self.pi.set_watchdog(self.gpio_1, 0)
self.in_code = False
if self.bits >= 26:
hex = self.get_hex()
if hex:
self.callback(self.bits, hex)
else:
logging.error("Wiegand receive error: Expected at least 26 got %i bits. raw: %s" %(self.bits, self.items))
except Exception as e:
logging.error("Wiegand callback error: " + str(e))
def button_cb(self, gpio_pin, level, tick):
print("button: gpio_pin:{}, level:{}, tick:{}".format(gpio_pin, level, tick))
def open_door(self):
self.pi.write(self.door_pin, 1)
sleep(3)
self.pi.write(self.door_pin, 0)
def cancel(self):
"""
Cancel the Wiegand decoder.
"""
self.cb_0.cancel()
self.cb_1.cancel()
self.button_cb_h.cancel()
self.pi.stop()

3
pyproject.toml Normal file
View File

@ -0,0 +1,3 @@
[build-system]
requires = ["setuptools >= 40.6.0", "wheel"]
build-backend = "setuptools.build_meta"

10
run.sh.example Executable file
View File

@ -0,0 +1,10 @@
#!/bin/sh
export KDOORPI_DOOR="name" # Door name
export KDOORPI_API_ALLOWED="http://server/allowed" # Used to fetch allowed cards
export KDOORPI_API_LONGPOLL="http://server/longpoll" # Open door now and force allowed cards sync endpoint
export KDOORPI_API_SWIPE="http://server/swipe" # On card swipe post log line here
export KDOORPI_API_KEY="keykeykey" # api key for accessing allowed and longpoll api endpoints
export KDOORPI_UID_SALT="saltsaltsalt" # shared slat used to hash card uid bytes
python3 -m kdoorpi

23
setup.py Normal file
View File

@ -0,0 +1,23 @@
from setuptools import find_packages, setup
setup(
name='kdoorpi',
version='0.0.0',
author="Arti Zirk",
author_email="arti@zirk.me",
description="K-Space Door client that talks to the hardware",
packages=find_packages(),
include_package_data=True,
zip_safe=False,
python_requires='>=3.5',
install_requires=["requests"],
extras_require={},
classifiers=[
'License :: OSI Approved :: MIT License',
'Programming Language :: Python :: 3',
'Programming Language :: Python :: 3 :: Only',
'Topic :: System :: Networking',
'Intended Audience :: System Administrators',
]
)