commit
e495bdac5f
22 changed files with 572 additions and 219 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
|
@ -103,6 +103,9 @@ ENV/
|
|||
# generated models
|
||||
/src/models
|
||||
|
||||
# snapshots
|
||||
/src/snapshots
|
||||
|
||||
# build files
|
||||
debian/howdy.substvars
|
||||
debian/files
|
||||
|
|
|
|||
|
|
@ -1,6 +1,11 @@
|
|||
sudo: required
|
||||
dist: xenial
|
||||
language: python
|
||||
python: "3.6"
|
||||
python:
|
||||
- "3.4"
|
||||
- "3.6"
|
||||
- "3.7"
|
||||
- "3.8-dev"
|
||||
|
||||
script:
|
||||
# Build the binary (.deb)
|
||||
|
|
|
|||
24
debian/changelog
vendored
24
debian/changelog
vendored
|
|
@ -1,3 +1,27 @@
|
|||
howdy (2.6.0) xenial; urgency=medium
|
||||
|
||||
* Added new options to capture a snapshot of failed or even successful logins
|
||||
* Added command that creates a new snapshot and saves it
|
||||
* Added version command
|
||||
* Added question to automatically set certainty value on installation
|
||||
* Added automatic logging to system-wide auth.log
|
||||
* Added clearer feedback when login is rejected due to dark frames (thanks @andrewmv!)
|
||||
* Refactored video capture logic (thanks @AnthonyWharton!)
|
||||
* Reordered the editor priorities for the config command
|
||||
* Fixed gstreamer warnings showing up in console (thanks @ajnart!)
|
||||
* Fixed issue where add command would never end
|
||||
* Fixed test command overlay not being in color (thanks @PetePriority!)
|
||||
* Fixed typo preventing timeout config option from working (thanks @Ajayneethikannan!)
|
||||
* Fixed old numpy installation failure (thanks @rushabh-v!)
|
||||
* Fixed issue where no PAM response would be returned
|
||||
* Fixed CLAHE not being applied equally to all video commands (thanks @PetePriority!)
|
||||
* Fixed an incorrect suggested command (thanks @TheButlah!)
|
||||
* Fixed missing release method in video capture class
|
||||
* Removed deprecated dlib flags (thanks @rhysperry111!)
|
||||
* Removed streamer as a required dependency
|
||||
|
||||
-- boltgolt <boltgolt@gmail.com> Mon, 22 Jun 2020 16:11:46 +0200
|
||||
|
||||
howdy (2.5.1) xenial; urgency=medium
|
||||
|
||||
* Removed dismiss_lockscreen as it could lock users out of their system (thanks @ujjwalbe, @ju916 and many others!)
|
||||
|
|
|
|||
4
debian/control
vendored
4
debian/control
vendored
|
|
@ -9,8 +9,8 @@ Vcs-Git: https://github.com/boltgolt/howdy
|
|||
Package: howdy
|
||||
Homepage: https://github.com/boltgolt/howdy
|
||||
Architecture: all
|
||||
Depends: ${misc:Depends}, curl|wget, python3, python3-pip, python3-dev, python3-setuptools, libpam-python, fswebcam, libopencv-dev, cmake, streamer
|
||||
Recommends: libatlas-base-dev | libopenblas-dev | liblapack-dev
|
||||
Depends: ${misc:Depends}, curl|wget, python3, python3-pip, python3-dev, python3-setuptools, libpam-python, libopencv-dev, cmake
|
||||
Recommends: libatlas-base-dev | libopenblas-dev | liblapack-dev, streamer
|
||||
Suggests: nvidia-cuda-dev (>= 7.5)
|
||||
Description: Howdy: Windows Hello style authentication for Linux.
|
||||
Use your built-in IR emitters and camera in combination with face recognition
|
||||
|
|
|
|||
12
debian/postinst
vendored
12
debian/postinst
vendored
|
|
@ -109,6 +109,12 @@ log("Upgrading pip to the latest version")
|
|||
# Update pip
|
||||
handleStatus(sc(["pip3", "install", "--upgrade", "pip"]))
|
||||
|
||||
|
||||
log("Upgrading numpy to the lateset version")
|
||||
|
||||
# Update numpy
|
||||
handleStatus(subprocess.call(["pip3", "install", "--upgrade", "numpy"]))
|
||||
|
||||
log("Downloading and unpacking data files")
|
||||
|
||||
# Run the bash script to download and unpack the .dat files needed
|
||||
|
|
@ -190,10 +196,14 @@ handleStatus(subprocess.call(["pip3", "install", "--no-cache-dir", "opencv-pytho
|
|||
|
||||
log("Configuring howdy")
|
||||
|
||||
campath = picked.split(";")[0]
|
||||
cert = picked.split(";")[1]
|
||||
|
||||
# Manually change the camera id to the one picked
|
||||
for line in fileinput.input(["/lib/security/howdy/config.ini"], inplace=1):
|
||||
line = line.replace("device_path = none", "device_path = " + picked)
|
||||
line = line.replace("device_path = none", "device_path = " + campath)
|
||||
line = line.replace("use_cnn = false", "use_cnn = " + str(cuda_used).lower())
|
||||
line = line.replace("certainty = 3.5", "certainty = " + cert)
|
||||
|
||||
print(line, end="")
|
||||
|
||||
|
|
|
|||
124
debian/preinst
vendored
124
debian/preinst
vendored
|
|
@ -7,6 +7,7 @@ def col(id):
|
|||
if id == 1: return "\033[32m"
|
||||
if id == 2: return "\033[33m"
|
||||
if id == 3: return "\033[31m"
|
||||
if id == 4: return "\033[1m"
|
||||
return "\033[0m"
|
||||
|
||||
|
||||
|
|
@ -38,7 +39,6 @@ if "install" not in sys.argv:
|
|||
# The default picked video device id
|
||||
picked = "none"
|
||||
|
||||
print(col(1) + "Starting IR camera check...\n" + col(0))
|
||||
|
||||
# If prompting has been disabled, skip camera check
|
||||
if "HOWDY_NO_PROMPT" in os.environ:
|
||||
|
|
@ -46,62 +46,104 @@ if "HOWDY_NO_PROMPT" in os.environ:
|
|||
|
||||
# Write the default device to disk and exit
|
||||
with open("/tmp/howdy_picked_device", "w") as out_file:
|
||||
out_file.write("none")
|
||||
out_file.write("none;3.5")
|
||||
|
||||
sys.exit(0)
|
||||
|
||||
# Get all devices
|
||||
devices = os.listdir("/dev/v4l/by-path")
|
||||
fscheck = subprocess.call(["which", "streamer"], stdout=subprocess.PIPE)
|
||||
|
||||
# Loop though all devices
|
||||
for dev in devices:
|
||||
time.sleep(.5)
|
||||
if fscheck == 1:
|
||||
print(col(2) + "\nWARNING: Could not automatically find the right webcam, manual configuration after installation required\n" + col(0))
|
||||
else:
|
||||
print(col(1) + "Starting IR camera check...\n" + col(0))
|
||||
|
||||
# The full path to the device is the default name
|
||||
device_name = "/dev/v4l/by-path/" + dev
|
||||
# Get the udevadm details to try to get a better name
|
||||
udevadm = subprocess.check_output(["udevadm info -r --query=all -n " + device_name], shell=True).decode("utf-8")
|
||||
# Get all devices
|
||||
devices = os.listdir("/dev/v4l/by-path")
|
||||
|
||||
# Loop though udevadm to search for a better name
|
||||
for line in udevadm.split("\n"):
|
||||
# Match it and encase it in quotes
|
||||
re_name = re.search('product.*=(.*)$', line, re.IGNORECASE)
|
||||
if re_name:
|
||||
device_name = '"' + re_name.group(1) + '"'
|
||||
# Loop though all devices
|
||||
for dev in devices:
|
||||
time.sleep(.5)
|
||||
|
||||
# Show what device we're using
|
||||
print("Trying " + device_name)
|
||||
# The full path to the device is the default name
|
||||
device_name = "/dev/v4l/by-path/" + dev
|
||||
# Get the udevadm details to try to get a better name
|
||||
udevadm = subprocess.check_output(["udevadm info -r --query=all -n " + device_name], shell=True).decode("utf-8")
|
||||
|
||||
# Let fswebcam keep the camera open in the background
|
||||
sub = subprocess.Popen(["streamer -t 1:0:0 -c /dev/v4l/by-path/" + dev + " -b 16 -f rgb24 -o /dev/null 1>/dev/null 2>/dev/null"], shell=True, preexec_fn=os.setsid)
|
||||
# Loop though udevadm to search for a better name
|
||||
for line in udevadm.split("\n"):
|
||||
# Match it and encase it in quotes
|
||||
re_name = re.search('product.*=(.*)$', line, re.IGNORECASE)
|
||||
if re_name:
|
||||
device_name = '"' + re_name.group(1) + '"'
|
||||
|
||||
try:
|
||||
# Ask the user if this is the right one
|
||||
print(col(2) + "One of your cameras should now be on." + col(0))
|
||||
ans = input("Did your IR emitters turn on? [y/N]: ")
|
||||
except KeyboardInterrupt:
|
||||
# Kill fswebcam if the user aborts
|
||||
# Show what device we're using
|
||||
print("Trying " + device_name)
|
||||
|
||||
# Let fswebcam keep the camera open in the background
|
||||
sub = subprocess.Popen(
|
||||
["streamer -t 1:0:0 -c /dev/v4l/by-path/" + dev + " -b 16 -f rgb24 -o /dev/null 1>/dev/null 2>/dev/null"],
|
||||
shell=True,
|
||||
preexec_fn=os.setsid,
|
||||
stdout=subprocess.PIPE,
|
||||
stdin=subprocess.PIPE)
|
||||
|
||||
try:
|
||||
# Ask the user if this is the right one
|
||||
print(col(2) + "One of your cameras should now be on." + col(0))
|
||||
ans = input("Did your IR emitters turn on? [y/N]: ")
|
||||
except KeyboardInterrupt:
|
||||
# Kill fswebcam if the user aborts
|
||||
os.killpg(os.getpgid(sub.pid), signal.SIGTERM)
|
||||
raise
|
||||
|
||||
# The user has answered, kill fswebcam
|
||||
os.killpg(os.getpgid(sub.pid), signal.SIGTERM)
|
||||
raise
|
||||
|
||||
# The user has answered, kill fswebcam
|
||||
os.killpg(os.getpgid(sub.pid), signal.SIGTERM)
|
||||
# Set this camera as picked if the answer was yes, go to the next one if no
|
||||
if ans.lower().strip() == "y" or ans.lower().strip() == "yes":
|
||||
picked = dev
|
||||
break
|
||||
else:
|
||||
print("Interpreting as a " + col(3) + "\"NO\"\n" + col(0))
|
||||
|
||||
# Set this camera as picked if the answer was yes, go to the next one if no
|
||||
if ans.lower().strip() == "y" or ans.lower().strip() == "yes":
|
||||
picked = dev
|
||||
break
|
||||
else:
|
||||
print("Interpreting as a " + col(3) + "\"NO\"\n" + col(0))
|
||||
# Abort if no camera was picked
|
||||
if picked == "none":
|
||||
print(col(3) + "No suitable IR camera found, aborting install." + col(0))
|
||||
sys.exit(23)
|
||||
|
||||
# Abort if no camera was picked
|
||||
if picked == "none":
|
||||
print(col(3) + "No suitable IR camera found, aborting install." + col(0))
|
||||
sys.exit(23)
|
||||
cert = 3.5
|
||||
|
||||
# Give time to read
|
||||
time.sleep(.5)
|
||||
|
||||
print(col(1) + "\nStarting certainty auto config..." + col(0))
|
||||
|
||||
# Give more time to read
|
||||
time.sleep(.5)
|
||||
|
||||
print("\n\nAfter detection, Howdy knows how certain it is that the match is correct.")
|
||||
print("How certain Howdy needs to be before authenticating you can be customized.")
|
||||
|
||||
print(col(4) + "\nF: Fast." + col(0))
|
||||
print("Allows more fuzzy matches, but speeds up the scanning process greatly.")
|
||||
print(col(4) + "\nB: Balanced." + col(0))
|
||||
print("Still relatively quick detection, but might not log you in when further away.")
|
||||
print(col(4) + "\nS: Secure." + col(0))
|
||||
print("The safest option, but will take much longer to authenticate you.")
|
||||
|
||||
print("\nYou can always change this setting in the config.")
|
||||
prof = input("What profile would you like to use? [f/b/s]: ")
|
||||
|
||||
if prof.lower().strip() == "f" or prof.lower().strip() == "fast":
|
||||
cert = 1.5
|
||||
elif prof.lower().strip() == "b" or prof.lower().strip() == "balanced":
|
||||
cert = 2.8
|
||||
elif prof.lower().strip() == "s" or prof.lower().strip() == "secure":
|
||||
cert = 4
|
||||
|
||||
# Write the result to disk so postinst can have a look at it
|
||||
with open("/tmp/howdy_picked_device", "w") as out_file:
|
||||
out_file.write("/dev/v4l/by-path/" + picked)
|
||||
out_file.write("/dev/v4l/by-path/" + picked + ";" + str(cert))
|
||||
|
||||
# Add a line break
|
||||
print("")
|
||||
|
|
|
|||
1
debian/source/options
vendored
1
debian/source/options
vendored
|
|
@ -2,6 +2,7 @@ tar-ignore = ".git"
|
|||
tar-ignore = ".gitignore"
|
||||
tar-ignore = ".github"
|
||||
tar-ignore = "models"
|
||||
tar-ignore = "snapshots"
|
||||
tar-ignore = "tests"
|
||||
tar-ignore = "README.md"
|
||||
tar-ignore = ".travis.yml"
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ Version: 2.5.1
|
|||
%if %{with_snapshot}
|
||||
Release: 0.1.git.%{date}%{shortcommit}%{?dist}
|
||||
%else
|
||||
Release: 3%{?dist}
|
||||
Release: 4%{?dist}
|
||||
%endif
|
||||
Summary: Windows Hello™ style authentication for Linux
|
||||
|
||||
|
|
@ -27,12 +27,9 @@ BuildRequires: polkit-devel
|
|||
%if 0%{?fedora}
|
||||
# We need python3-devel for pathfix.py
|
||||
BuildRequires: python3-devel
|
||||
Requires: python3dist(dlib) >= 6.0
|
||||
Requires: python3dist(v4l2)
|
||||
Requires: python3-face_recognition
|
||||
Supplements: python3-face_recognition_models
|
||||
Requires: python3dist(dlib) >= 6.0
|
||||
Requires: python3-opencv
|
||||
Requires: python3-pam
|
||||
Requires: pam_python
|
||||
%endif
|
||||
|
||||
|
||||
|
|
|
|||
54
src/cli.py
54
src/cli.py
|
|
@ -26,38 +26,44 @@ if user == "root" or user is None:
|
|||
user = env_user
|
||||
|
||||
# Basic command setup
|
||||
parser = argparse.ArgumentParser(description="Command line interface for Howdy face authentication.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
add_help=False,
|
||||
prog="howdy",
|
||||
epilog="For support please visit\nhttps://github.com/boltgolt/howdy")
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Command line interface for Howdy face authentication.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
add_help=False,
|
||||
prog="howdy",
|
||||
epilog="For support please visit\nhttps://github.com/boltgolt/howdy")
|
||||
|
||||
# Add an argument for the command
|
||||
parser.add_argument("command",
|
||||
help="The command option to execute, can be one of the following: add, clear, config, disable, list, remove or test.",
|
||||
metavar="command",
|
||||
choices=["add", "clear", "config", "disable", "list", "remove", "test"])
|
||||
parser.add_argument(
|
||||
"command",
|
||||
help="The command option to execute, can be one of the following: add, clear, config, disable, list, remove, snapshot, test or version.",
|
||||
metavar="command",
|
||||
choices=["add", "clear", "config", "disable", "list", "remove", "snapshot", "test", "version"])
|
||||
|
||||
# Add an argument for the extra arguments of diable and remove
|
||||
parser.add_argument("argument",
|
||||
help="Either 0 (enable) or 1 (disable) for the disable command, or the model ID for the remove command.",
|
||||
nargs="?")
|
||||
parser.add_argument(
|
||||
"argument",
|
||||
help="Either 0 (enable) or 1 (disable) for the disable command, or the model ID for the remove command.",
|
||||
nargs="?")
|
||||
|
||||
# Add the user flag
|
||||
parser.add_argument("-U", "--user",
|
||||
default=user,
|
||||
help="Set the user account to use.")
|
||||
parser.add_argument(
|
||||
"-U", "--user",
|
||||
default=user,
|
||||
help="Set the user account to use.")
|
||||
|
||||
# Add the -y flag
|
||||
parser.add_argument("-y",
|
||||
help="Skip all questions.",
|
||||
action="store_true")
|
||||
parser.add_argument(
|
||||
"-y",
|
||||
help="Skip all questions.",
|
||||
action="store_true")
|
||||
|
||||
# Overwrite the default help message so we can use a uppercase S
|
||||
parser.add_argument("-h", "--help",
|
||||
action="help",
|
||||
default=argparse.SUPPRESS,
|
||||
help="Show this help message and exit.")
|
||||
parser.add_argument(
|
||||
"-h", "--help",
|
||||
action="help",
|
||||
default=argparse.SUPPRESS,
|
||||
help="Show this help message and exit.")
|
||||
|
||||
# If we only have 1 argument we print the help text
|
||||
if len(sys.argv) < 2:
|
||||
|
|
@ -97,5 +103,9 @@ elif args.command == "list":
|
|||
import cli.list
|
||||
elif args.command == "remove":
|
||||
import cli.remove
|
||||
elif args.command == "snapshot":
|
||||
import cli.snap
|
||||
elif args.command == "test":
|
||||
import cli.test
|
||||
else:
|
||||
print("Howdy 2.6.0")
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import configparser
|
|||
import builtins
|
||||
import cv2
|
||||
import numpy as np
|
||||
from recorders.video_capture import VideoCapture
|
||||
|
||||
# Try to import dlib and give a nice error if we can't
|
||||
# Add should be the first point where import issues show up
|
||||
|
|
@ -35,10 +36,6 @@ if not os.path.isfile(path + "/../dlib-data/shape_predictor_5_face_landmarks.dat
|
|||
config = configparser.ConfigParser()
|
||||
config.read(path + "/../config.ini")
|
||||
|
||||
if not os.path.exists(config.get("video", "device_path")):
|
||||
print("Camera path is not configured correctly, please edit the 'device_path' config value.")
|
||||
sys.exit(1)
|
||||
|
||||
use_cnn = config.getboolean("core", "use_cnn", fallback=False)
|
||||
if use_cnn:
|
||||
face_detector = dlib.cnn_face_detection_model_v1(path + "/../dlib-data/mmod_human_face_detector.dat")
|
||||
|
|
@ -98,35 +95,8 @@ insert_model = {
|
|||
"data": []
|
||||
}
|
||||
|
||||
# Check if the user explicitly set ffmpeg as recorder
|
||||
if config.get("video", "recording_plugin") == "ffmpeg":
|
||||
# Set the capture source for ffmpeg
|
||||
from recorders.ffmpeg_reader import ffmpeg_reader
|
||||
video_capture = ffmpeg_reader(config.get("video", "device_path"), config.get("video", "device_format"))
|
||||
elif config.get("video", "recording_plugin") == "pyv4l2":
|
||||
# Set the capture source for pyv4l2
|
||||
from recorders.pyv4l2_reader import pyv4l2_reader
|
||||
video_capture = pyv4l2_reader(config.get("video", "device_path"), config.get("video", "device_format"))
|
||||
else:
|
||||
# Start video capture on the IR camera through OpenCV
|
||||
video_capture = cv2.VideoCapture(config.get("video", "device_path"), cv2.CAP_V4L)
|
||||
|
||||
# Force MJPEG decoding if true
|
||||
if config.getboolean("video", "force_mjpeg", fallback=False):
|
||||
# Set a magic number, will enable MJPEG but is badly documentated
|
||||
video_capture.set(cv2.CAP_PROP_FOURCC, 1196444237)
|
||||
|
||||
# Set the frame width and height if requested
|
||||
fw = config.getint("video", "frame_width", fallback=-1)
|
||||
fh = config.getint("video", "frame_height", fallback=-1)
|
||||
if fw != -1:
|
||||
video_capture.set(cv2.CAP_PROP_FRAME_WIDTH, fw)
|
||||
|
||||
if fh != -1:
|
||||
video_capture.set(cv2.CAP_PROP_FRAME_HEIGHT, fh)
|
||||
|
||||
# Request a frame to wake the camera up
|
||||
video_capture.grab()
|
||||
# Set up video_capture
|
||||
video_capture = VideoCapture(config)
|
||||
|
||||
print("\nPlease look straight into the camera")
|
||||
|
||||
|
|
@ -135,28 +105,51 @@ time.sleep(2)
|
|||
|
||||
# Will contain found face encodings
|
||||
enc = []
|
||||
# Count the amount or read frames
|
||||
# Count the number of read frames
|
||||
frames = 0
|
||||
# Count the number of illuminated read frames
|
||||
valid_frames = 0
|
||||
# Count the number of illuminated frames that
|
||||
# were rejected for being too dark
|
||||
dark_tries = 0
|
||||
# Track the running darkness total
|
||||
dark_running_total = 0
|
||||
face_locations = None
|
||||
|
||||
dark_threshold = config.getfloat("video", "dark_threshold")
|
||||
|
||||
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
|
||||
|
||||
# Loop through frames till we hit a timeout
|
||||
while frames < 60:
|
||||
frames += 1
|
||||
# Grab a single frame of video
|
||||
# Don't remove ret, it doesn't work without it
|
||||
ret, frame = video_capture.read()
|
||||
frame, gsframe = video_capture.read_frame()
|
||||
gsframe = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
gsframe = clahe.apply(gsframe)
|
||||
|
||||
# Create a histogram of the image with 8 values
|
||||
hist = cv2.calcHist([gsframe], [0], None, [8], [0, 256])
|
||||
# All values combined for percentage calculation
|
||||
hist_total = np.sum(hist)
|
||||
|
||||
# If the image is fully black or the frame exceeds threshold,
|
||||
# Calculate frame darkness
|
||||
darkness = (hist[0] / hist_total * 100)
|
||||
|
||||
# If the image is fully black due to a bad camera read,
|
||||
# skip to the next frame
|
||||
if hist_total == 0 or (hist[0] / hist_total * 100 > dark_threshold):
|
||||
if (hist_total == 0) or (darkness == 100):
|
||||
continue
|
||||
|
||||
frames += 1
|
||||
# Include this frame in calculating our average session brightness
|
||||
dark_running_total += darkness
|
||||
valid_frames += 1
|
||||
|
||||
# If the image exceeds darkness threshold due to subject distance,
|
||||
# skip to the next frame
|
||||
if (darkness > dark_threshold):
|
||||
dark_tries += 1
|
||||
continue
|
||||
|
||||
# Get all faces from that frame as encodings
|
||||
face_locations = face_detector(gsframe, 1)
|
||||
|
|
@ -167,12 +160,20 @@ while frames < 60:
|
|||
|
||||
video_capture.release()
|
||||
|
||||
# If more than 1 faces are detected we can't know wich one belongs to the user
|
||||
if len(face_locations) > 1:
|
||||
print("Multiple faces detected, aborting")
|
||||
# If we've found no faces, try to determine why
|
||||
if face_locations is None or not face_locations:
|
||||
if valid_frames == 0:
|
||||
print("Camera saw only black frames - is IR emitter working?")
|
||||
elif valid_frames == dark_tries:
|
||||
print("All frames were too dark, please check dark_threshold in config")
|
||||
print("Average darkness: " + str(dark_running_total / valid_frames) + ", Threshold: " + str(dark_threshold))
|
||||
else:
|
||||
print("No face detected, aborting")
|
||||
sys.exit(1)
|
||||
elif not face_locations:
|
||||
print("No face detected, aborting")
|
||||
|
||||
# If more than 1 faces are detected we can't know wich one belongs to the user
|
||||
elif len(face_locations) > 1:
|
||||
print("Multiple faces detected, aborting")
|
||||
sys.exit(1)
|
||||
|
||||
face_location = face_locations[0]
|
||||
|
|
|
|||
|
|
@ -11,10 +11,10 @@ print("Opening config.ini in the default editor")
|
|||
editor = "/bin/nano"
|
||||
|
||||
# Use the user preferred editor if available
|
||||
if os.path.isfile("/etc/alternatives/editor"):
|
||||
editor = "/etc/alternatives/editor"
|
||||
elif "EDITOR" in os.environ:
|
||||
if "EDITOR" in os.environ:
|
||||
editor = os.environ["EDITOR"]
|
||||
elif os.path.isfile("/etc/alternatives/editor"):
|
||||
editor = "/etc/alternatives/editor"
|
||||
|
||||
# Open the editor as a subprocess and fork it
|
||||
subprocess.call([editor, os.path.dirname(os.path.realpath(__file__)) + "/../config.ini"])
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@
|
|||
# Import required modules
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
import builtins
|
||||
import fileinput
|
||||
import configparser
|
||||
|
|
|
|||
|
|
@ -14,7 +14,7 @@ user = builtins.howdy_user
|
|||
# Check if the models file has been created yet
|
||||
if not os.path.exists(path + "/models"):
|
||||
print("Face models have not been initialized yet, please run:")
|
||||
print("\n\thowdy " + user + " add\n")
|
||||
print("\n\tsudo howdy -U " + user + " add\n")
|
||||
sys.exit(1)
|
||||
|
||||
# Path to the models file
|
||||
|
|
@ -25,7 +25,7 @@ try:
|
|||
encodings = json.load(open(enc_file))
|
||||
except FileNotFoundError:
|
||||
print("No face model known for the user " + user + ", please run:")
|
||||
print("\n\thowdy " + user + " add\n")
|
||||
print("\n\tsudo howdy -U " + user + " add\n")
|
||||
sys.exit(1)
|
||||
|
||||
# Print a header
|
||||
|
|
|
|||
51
src/cli/snap.py
Normal file
51
src/cli/snap.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
# Create a snapshot
|
||||
|
||||
# Import required modules
|
||||
import os
|
||||
import configparser
|
||||
import datetime
|
||||
import snapshot
|
||||
from recorders.video_capture import VideoCapture
|
||||
|
||||
# Get the absolute path to the current directory
|
||||
path = os.path.abspath(__file__ + "/..")
|
||||
|
||||
# Read the config
|
||||
config = configparser.ConfigParser()
|
||||
config.read(path + "/../config.ini")
|
||||
|
||||
# Start video capture
|
||||
video_capture = VideoCapture(config)
|
||||
|
||||
# Read a frame to activate emitters
|
||||
video_capture.read_frame()
|
||||
|
||||
# Read exposure and dark_thresholds from config to use in the main loop
|
||||
exposure = config.getint("video", "exposure", fallback=-1)
|
||||
dark_threshold = config.getfloat("video", "dark_threshold")
|
||||
|
||||
# COllection of recorded frames
|
||||
frames = []
|
||||
|
||||
while True:
|
||||
# Grab a single frame of video
|
||||
frame, gsframe = video_capture.read_frame()
|
||||
|
||||
# Add the frame to the list
|
||||
frames.append(frame)
|
||||
|
||||
# Stop the loop if we have 4 frames
|
||||
if len(frames) >= 4:
|
||||
break
|
||||
|
||||
# Generate a snapshot image from the frames
|
||||
file = snapshot.generate(frames, [
|
||||
"GENERATED SNAPSHOT",
|
||||
"Date: " + datetime.datetime.utcnow().strftime("%Y/%m/%d %H:%M:%S UTC"),
|
||||
"Dark threshold config: " + str(config.getfloat("video", "dark_threshold", fallback=50.0)),
|
||||
"Certainty config: " + str(config.getfloat("video", "certainty", fallback=3.5))
|
||||
])
|
||||
|
||||
# Show the file location in console
|
||||
print("Generated snapshot saved as")
|
||||
print(file)
|
||||
|
|
@ -7,6 +7,7 @@ import sys
|
|||
import time
|
||||
import cv2
|
||||
import dlib
|
||||
from recorders.video_capture import VideoCapture
|
||||
|
||||
# Get the absolute path to the current file
|
||||
path = os.path.dirname(os.path.abspath(__file__))
|
||||
|
|
@ -20,22 +21,7 @@ if config.get("video", "recording_plugin") != "opencv":
|
|||
print("Aborting")
|
||||
sys.exit(12)
|
||||
|
||||
# Start capturing from the configured webcam
|
||||
video_capture = cv2.VideoCapture(config.get("video", "device_path"), cv2.CAP_V4L)
|
||||
|
||||
# Force MJPEG decoding if true
|
||||
if config.getboolean("video", "force_mjpeg", fallback=False):
|
||||
# Set a magic number, will enable MJPEG but is badly documented
|
||||
video_capture.set(cv2.CAP_PROP_FOURCC, 1196444237)
|
||||
|
||||
# Set the frame width and height if requested
|
||||
fw = config.getint("video", "frame_width", fallback=-1)
|
||||
fh = config.getint("video", "frame_height", fallback=-1)
|
||||
if fw != -1:
|
||||
video_capture.set(cv2.CAP_PROP_FRAME_WIDTH, fw)
|
||||
|
||||
if fh != -1:
|
||||
video_capture.set(cv2.CAP_PROP_FRAME_HEIGHT, fh)
|
||||
video_capture = VideoCapture(config)
|
||||
|
||||
# Read exposure and dark_thresholds from config to use in the main loop
|
||||
exposure = config.getint("video", "exposure", fallback=-1)
|
||||
|
|
@ -63,7 +49,9 @@ def print_text(line_number, text):
|
|||
"""Print the status text by line number"""
|
||||
cv2.putText(overlay, text, (10, height - 10 - (10 * line_number)), cv2.FONT_HERSHEY_SIMPLEX, .3, (0, 255, 0), 0, cv2.LINE_AA)
|
||||
|
||||
|
||||
use_cnn = config.getboolean('core', 'use_cnn', fallback=False)
|
||||
|
||||
if use_cnn:
|
||||
face_detector = dlib.cnn_face_detection_model_v1(
|
||||
path + '/../dlib-data/mmod_human_face_detector.dat'
|
||||
|
|
@ -109,21 +97,12 @@ try:
|
|||
sec_frames = 0
|
||||
|
||||
# Grab a single frame of video
|
||||
ret, frame = video_capture.read()
|
||||
|
||||
try:
|
||||
# Convert from color to grayscale
|
||||
# First processing of frame, so frame errors show up here
|
||||
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
except RuntimeError:
|
||||
pass
|
||||
except cv2.error:
|
||||
print("\nUnknown camera, please check your 'device_path' config value.\n")
|
||||
raise
|
||||
_, frame = video_capture.read_frame()
|
||||
|
||||
frame = clahe.apply(frame)
|
||||
# Make a frame to put overlays in
|
||||
overlay = frame.copy()
|
||||
overlay = cv2.cvtColor(overlay, cv2.COLOR_GRAY2BGR)
|
||||
|
||||
# Fetch the frame height and width
|
||||
height, width = frame.shape[:2]
|
||||
|
|
@ -190,6 +169,7 @@ try:
|
|||
|
||||
# Add the overlay to the frame with some transparency
|
||||
alpha = 0.65
|
||||
frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
|
||||
cv2.addWeighted(overlay, alpha, frame, 1 - alpha, 0, frame)
|
||||
|
||||
# Show the image in a window
|
||||
|
|
@ -211,8 +191,8 @@ try:
|
|||
# are captured and even after a delay it does not
|
||||
# always work. Setting exposure at every frame is
|
||||
# reliable though.
|
||||
video_capture.set(cv2.CAP_PROP_AUTO_EXPOSURE, 1.0) # 1 = Manual
|
||||
video_capture.set(cv2.CAP_PROP_EXPOSURE, float(exposure))
|
||||
video_capture.intenal.set(cv2.CAP_PROP_AUTO_EXPOSURE, 1.0) # 1 = Manual
|
||||
video_capture.intenal.set(cv2.CAP_PROP_EXPOSURE, float(exposure))
|
||||
|
||||
# On ctrl+C
|
||||
except KeyboardInterrupt:
|
||||
|
|
@ -220,5 +200,4 @@ except KeyboardInterrupt:
|
|||
print("\nClosing window")
|
||||
|
||||
# Release handle to the webcam
|
||||
video_capture.release()
|
||||
cv2.destroyAllWindows()
|
||||
|
|
|
|||
136
src/compare.py
136
src/compare.py
|
|
@ -16,8 +16,11 @@ import json
|
|||
import configparser
|
||||
import cv2
|
||||
import dlib
|
||||
import datetime
|
||||
import snapshot
|
||||
import numpy as np
|
||||
import _thread as thread
|
||||
from recorders.video_capture import VideoCapture
|
||||
|
||||
|
||||
def init_detector(lock):
|
||||
|
|
@ -47,10 +50,16 @@ def init_detector(lock):
|
|||
lock.release()
|
||||
|
||||
|
||||
def stop(status):
|
||||
"""Stop the execution and close video stream"""
|
||||
video_capture.release()
|
||||
sys.exit(status)
|
||||
def make_snapshot(type):
|
||||
"""Generate snapshot after detection"""
|
||||
snapshot.generate(snapframes, [
|
||||
type + " LOGIN",
|
||||
"Date: " + datetime.datetime.utcnow().strftime("%Y/%m/%d %H:%M:%S UTC"),
|
||||
"Scan time: " + str(round(time.time() - timings["fr"], 2)) + "s",
|
||||
"Frames: " + str(frames) + " (" + str(round(frames / (time.time() - timings["fr"]), 2)) + "FPS)",
|
||||
"Hostname: " + os.uname().nodename,
|
||||
"Best certainty value: " + str(round(lowest_certainty * 10, 1))
|
||||
])
|
||||
|
||||
|
||||
# Make sure we were given an username to tast against
|
||||
|
|
@ -66,11 +75,17 @@ user = sys.argv[1]
|
|||
models = []
|
||||
# Encoded face models
|
||||
encodings = []
|
||||
# Amount of ignored 100% black frames
|
||||
black_tries = 0
|
||||
# Amount of ingnored dark frames
|
||||
dark_tries = 0
|
||||
# Total amount of frames captured
|
||||
frames = 0
|
||||
# face recognition/detection instances
|
||||
# Captured frames for snapshot capture
|
||||
snapframes = []
|
||||
# Tracks the lowest certainty value in the loop
|
||||
lowest_certainty = 10
|
||||
# Face recognition/detection instances
|
||||
face_detector = None
|
||||
pose_predictor = None
|
||||
face_encoder = None
|
||||
|
|
@ -94,10 +109,12 @@ config.read(PATH + "/config.ini")
|
|||
|
||||
# Get all config values needed
|
||||
use_cnn = config.getboolean("core", "use_cnn", fallback=False)
|
||||
timeout = config.getint("video", "timout", fallback=5)
|
||||
timeout = config.getint("video", "timeout", fallback=5)
|
||||
dark_threshold = config.getfloat("video", "dark_threshold", fallback=50.0)
|
||||
video_certainty = config.getfloat("video", "certainty", fallback=3.5) / 10
|
||||
end_report = config.getboolean("debug", "end_report", fallback=False)
|
||||
capture_failed = config.getboolean("snapshots", "capture_failed", fallback=False)
|
||||
capture_successful = config.getboolean("snapshots", "capture_successful", fallback=False)
|
||||
|
||||
# Save the time needed to start the script
|
||||
timings["in"] = time.time() - timings["st"]
|
||||
|
|
@ -113,36 +130,7 @@ thread.start_new_thread(init_detector, (lock, ))
|
|||
# Start video capture on the IR camera
|
||||
timings["ic"] = time.time()
|
||||
|
||||
# Check if the user explicitly set ffmpeg as recorder
|
||||
if config.get("video", "recording_plugin") == "ffmpeg":
|
||||
# Set the capture source for ffmpeg
|
||||
from recorders.ffmpeg_reader import ffmpeg_reader
|
||||
video_capture = ffmpeg_reader(config.get("video", "device_path"), config.get("video", "device_format"))
|
||||
elif config.get("video", "recording_plugin") == "pyv4l2":
|
||||
# Set the capture source for pyv4l2
|
||||
from recorders.pyv4l2_reader import pyv4l2_reader
|
||||
video_capture = pyv4l2_reader(config.get("video", "device_path"), config.get("video", "device_format"))
|
||||
else:
|
||||
# Start video capture on the IR camera through OpenCV
|
||||
video_capture = cv2.VideoCapture(config.get("video", "device_path"), cv2.CAP_V4L)
|
||||
|
||||
# Force MJPEG decoding if true
|
||||
if config.getboolean("video", "force_mjpeg", fallback=False):
|
||||
# Set a magic number, will enable MJPEG but is badly documented
|
||||
# 1196444237 is "GPJM" in ASCII
|
||||
video_capture.set(cv2.CAP_PROP_FOURCC, 1196444237)
|
||||
|
||||
# Set the frame width and height if requested
|
||||
fw = config.getint("video", "frame_width", fallback=-1)
|
||||
fh = config.getint("video", "frame_height", fallback=-1)
|
||||
if fw != -1:
|
||||
video_capture.set(cv2.CAP_PROP_FRAME_WIDTH, fw)
|
||||
if fh != -1:
|
||||
video_capture.set(cv2.CAP_PROP_FRAME_HEIGHT, fh)
|
||||
|
||||
# Capture a single frame so the camera becomes active
|
||||
# This will let the camera adjust its light levels while we're importing for faster scanning
|
||||
video_capture.grab()
|
||||
video_capture = VideoCapture(config)
|
||||
|
||||
# Read exposure from config to use in the main loop
|
||||
exposure = config.getint("video", "exposure", fallback=-1)
|
||||
|
|
@ -158,7 +146,7 @@ del lock
|
|||
# Fetch the max frame height
|
||||
max_height = config.getfloat("video", "max_height", fallback=0.0)
|
||||
# Get the height of the image
|
||||
height = video_capture.get(cv2.CAP_PROP_FRAME_HEIGHT) or 1
|
||||
height = video_capture.internal.get(cv2.CAP_PROP_FRAME_HEIGHT) or 1
|
||||
|
||||
# Calculate the amount the image has to shrink
|
||||
scaling_factor = (max_height / height) or 1
|
||||
|
|
@ -170,7 +158,11 @@ end_report = config.getboolean("debug", "end_report")
|
|||
|
||||
# Start the read loop
|
||||
frames = 0
|
||||
valid_frames = 0
|
||||
timings["fr"] = time.time()
|
||||
dark_running_total = 0
|
||||
|
||||
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
|
||||
|
||||
while True:
|
||||
# Increment the frame count every loop
|
||||
|
|
@ -178,33 +170,46 @@ while True:
|
|||
|
||||
# Stop if we've exceded the time limit
|
||||
if time.time() - timings["fr"] > timeout:
|
||||
stop(11)
|
||||
# Create a timeout snapshot if enabled
|
||||
if capture_failed:
|
||||
make_snapshot("FAILED")
|
||||
|
||||
if dark_tries == valid_frames:
|
||||
print("All frames were too dark, please check dark_threshold in config")
|
||||
print("Average darkness: " + str(dark_running_total / valid_frames) + ", Threshold: " + str(dark_threshold))
|
||||
sys.exit(13)
|
||||
else:
|
||||
sys.exit(11)
|
||||
|
||||
# Grab a single frame of video
|
||||
ret, frame = video_capture.read()
|
||||
frame, gsframe = video_capture.read_frame()
|
||||
gsframe = clahe.apply(gsframe)
|
||||
|
||||
if frames == 1 and ret is False:
|
||||
print("Could not read from camera")
|
||||
exit(12)
|
||||
|
||||
try:
|
||||
# Convert from color to grayscale
|
||||
# First processing of frame, so frame errors show up here
|
||||
gsframe = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
except RuntimeError:
|
||||
gsframe = frame
|
||||
except cv2.error:
|
||||
print("\nUnknown camera, please check your 'device_path' config value.\n")
|
||||
raise
|
||||
# If snapshots have been turned on
|
||||
if capture_failed or capture_successful:
|
||||
# Start capturing frames for the snapshot
|
||||
if len(snapframes) < 3:
|
||||
snapframes.append(frame)
|
||||
|
||||
# Create a histogram of the image with 8 values
|
||||
hist = cv2.calcHist([gsframe], [0], None, [8], [0, 256])
|
||||
# All values combined for percentage calculation
|
||||
hist_total = np.sum(hist)
|
||||
|
||||
# If the image is fully black or the frame exceeds threshold,
|
||||
# Calculate frame darkness
|
||||
darkness = (hist[0] / hist_total * 100)
|
||||
|
||||
# If the image is fully black due to a bad camera read,
|
||||
# skip to the next frame
|
||||
if hist_total == 0 or (hist[0] / hist_total * 100 > dark_threshold):
|
||||
if (hist_total == 0) or (darkness == 100):
|
||||
black_tries += 1
|
||||
continue
|
||||
|
||||
dark_running_total += darkness
|
||||
valid_frames += 1
|
||||
# If the image exceeds darkness threshold due to subject distance,
|
||||
# skip to the next frame
|
||||
if (darkness > dark_threshold):
|
||||
dark_tries += 1
|
||||
continue
|
||||
|
||||
|
|
@ -234,10 +239,14 @@ while True:
|
|||
match_index = np.argmin(matches)
|
||||
match = matches[match_index]
|
||||
|
||||
# Update certainty if we have a new low
|
||||
if lowest_certainty > match:
|
||||
lowest_certainty = match
|
||||
|
||||
# Check if a match that's confident enough
|
||||
if 0 < match < video_certainty:
|
||||
timings["tt"] = time.time() - timings["st"]
|
||||
timings["fr"] = time.time() - timings["fr"]
|
||||
timings["fl"] = time.time() - timings["fr"]
|
||||
|
||||
# If set to true in the config, print debug text
|
||||
if end_report:
|
||||
|
|
@ -251,25 +260,30 @@ while True:
|
|||
print(" Open cam + load libs: %dms" % (round(max(timings["ll"], timings["ic"]) * 1000, )))
|
||||
print_timing(" Opening the camera", "ic")
|
||||
print_timing(" Importing recognition libs", "ll")
|
||||
print_timing("Searching for known face", "fr")
|
||||
print_timing("Searching for known face", "fl")
|
||||
print_timing("Total time", "tt")
|
||||
|
||||
print("\nResolution")
|
||||
width = video_capture.get(cv2.CAP_PROP_FRAME_WIDTH) or 1
|
||||
width = video_capture.fw or 1
|
||||
print(" Native: %dx%d" % (height, width))
|
||||
# Save the new size for diagnostics
|
||||
scale_height, scale_width = frame.shape[:2]
|
||||
print(" Used: %dx%d" % (scale_height, scale_width))
|
||||
|
||||
# Show the total number of frames and calculate the FPS by deviding it by the total scan time
|
||||
print("\nFrames searched: %d (%.2f fps)" % (frames, frames / timings["fr"]))
|
||||
print("\nFrames searched: %d (%.2f fps)" % (frames, frames / timings["fl"]))
|
||||
print("Black frames ignored: %d " % (black_tries, ))
|
||||
print("Dark frames ignored: %d " % (dark_tries, ))
|
||||
print("Certainty of winning frame: %.3f" % (match * 10, ))
|
||||
|
||||
print("Winning model: %d (\"%s\")" % (match_index, models[match_index]["label"]))
|
||||
|
||||
# Make snapshot if enabled
|
||||
if capture_successful:
|
||||
make_snapshot("SUCCESSFUL")
|
||||
|
||||
# End peacefully
|
||||
stop(0)
|
||||
sys.exit(0)
|
||||
|
||||
if exposure != -1:
|
||||
# For a strange reason on some cameras (e.g. Lenoxo X1E)
|
||||
|
|
@ -277,5 +291,5 @@ while True:
|
|||
# are captured and even after a delay it does not
|
||||
# always work. Setting exposure at every frame is
|
||||
# reliable though.
|
||||
video_capture.set(cv2.CAP_PROP_AUTO_EXPOSURE, 1.0) # 1 = Manual
|
||||
video_capture.set(cv2.CAP_PROP_EXPOSURE, float(exposure))
|
||||
video_capture.intenal.set(cv2.CAP_PROP_AUTO_EXPOSURE, 1.0) # 1 = Manual
|
||||
video_capture.intenal.set(cv2.CAP_PROP_EXPOSURE, float(exposure))
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ use_cnn = false
|
|||
[video]
|
||||
# The certainty of the detected face belonging to the user of the account
|
||||
# On a scale from 1 to 10, values above 5 are not recommended
|
||||
# Lower is better
|
||||
certainty = 3.5
|
||||
|
||||
# The number of seconds to search before timing out
|
||||
|
|
@ -73,6 +74,14 @@ force_mjpeg = false
|
|||
# OPENCV only.
|
||||
exposure = -1
|
||||
|
||||
[snapshots]
|
||||
# Capture snapshots of failed login attempts and save them to disk with metadata
|
||||
# Snapshots are saved to the "snapshots" folder
|
||||
capture_failed = true
|
||||
|
||||
# Do the same as the option above but for successful attempts
|
||||
capture_successful = true
|
||||
|
||||
[debug]
|
||||
# Show a short but detailed diagnostic report in console
|
||||
# Enabling this can cause some UI apps to fail, only enable it to debug
|
||||
|
|
|
|||
|
|
@ -22,4 +22,4 @@ fi
|
|||
|
||||
# Uncompress the data files and delete the original archive
|
||||
echo "Unpacking..."
|
||||
bzip2 -d *.bz2
|
||||
bzip2 -d -f *.bz2
|
||||
|
|
|
|||
BIN
src/logo.png
Normal file
BIN
src/logo.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 3 KiB |
24
src/pam.py
24
src/pam.py
|
|
@ -2,9 +2,9 @@
|
|||
|
||||
# Import required modules
|
||||
import subprocess
|
||||
import sys
|
||||
import os
|
||||
import glob
|
||||
import syslog
|
||||
|
||||
# pam-python is running python 2, so we use the old module here
|
||||
import ConfigParser
|
||||
|
|
@ -19,22 +19,24 @@ def doAuth(pamh):
|
|||
|
||||
# Abort is Howdy is disabled
|
||||
if config.getboolean("core", "disabled"):
|
||||
sys.exit(0)
|
||||
return pamh.PAM_AUTHINFO_UNAVAIL
|
||||
|
||||
# Abort if we're in a remote SSH env
|
||||
if config.getboolean("core", "ignore_ssh"):
|
||||
if "SSH_CONNECTION" in os.environ or "SSH_CLIENT" in os.environ or "SSHD_OPTS" in os.environ:
|
||||
sys.exit(0)
|
||||
return pamh.PAM_AUTHINFO_UNAVAIL
|
||||
|
||||
# Abort if lid is closed
|
||||
if config.getboolean("core", "ignore_closed_lid"):
|
||||
if any("closed" in open(f).read() for f in glob.glob("/proc/acpi/button/lid/*/state")):
|
||||
sys.exit(0)
|
||||
return pamh.PAM_AUTHINFO_UNAVAIL
|
||||
|
||||
# Alert the user that we are doing face detection
|
||||
if config.getboolean("core", "detection_notice"):
|
||||
pamh.conversation(pamh.Message(pamh.PAM_TEXT_INFO, "Attempting face detection"))
|
||||
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Attempting facial authentication for user " + pamh.get_user())
|
||||
|
||||
# Run compare as python3 subprocess to circumvent python version and import issues
|
||||
status = subprocess.call(["/usr/bin/python3", os.path.dirname(os.path.abspath(__file__)) + "/compare.py", pamh.get_user()])
|
||||
|
||||
|
|
@ -42,13 +44,25 @@ def doAuth(pamh):
|
|||
if status == 10:
|
||||
if not config.getboolean("core", "suppress_unknown"):
|
||||
pamh.conversation(pamh.Message(pamh.PAM_ERROR_MSG, "No face model known"))
|
||||
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Failure, no face model known")
|
||||
return pamh.PAM_USER_UNKNOWN
|
||||
|
||||
# Status 11 means we exceded the maximum retry count
|
||||
elif status == 11:
|
||||
pamh.conversation(pamh.Message(pamh.PAM_ERROR_MSG, "Face detection timeout reached"))
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Failure, timeout reached")
|
||||
return pamh.PAM_AUTH_ERR
|
||||
|
||||
# Status 12 means we aborted
|
||||
elif status == 12:
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Failure, general abort")
|
||||
return pamh.PAM_AUTH_ERR
|
||||
|
||||
# Status 13 means the image was too dark
|
||||
elif status == 13:
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Failure, image too dark")
|
||||
pamh.conversation(pamh.Message(pamh.PAM_ERROR_MSG, "Face detection image too dark"))
|
||||
return pamh.PAM_AUTH_ERR
|
||||
# Status 0 is a successful exit
|
||||
elif status == 0:
|
||||
|
|
@ -56,10 +70,12 @@ def doAuth(pamh):
|
|||
if not config.getboolean("core", "no_confirmation"):
|
||||
pamh.conversation(pamh.Message(pamh.PAM_TEXT_INFO, "Identified face as " + pamh.get_user()))
|
||||
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Login approved")
|
||||
return pamh.PAM_SUCCESS
|
||||
|
||||
# Otherwise, we can't discribe what happend but it wasn't successful
|
||||
pamh.conversation(pamh.Message(pamh.PAM_ERROR_MSG, "Unknown error: " + str(status)))
|
||||
syslog.syslog(syslog.LOG_AUTH, "[HOWDY] Failure, unknown error" + str(status))
|
||||
return pamh.PAM_SYSTEM_ERR
|
||||
|
||||
|
||||
|
|
|
|||
130
src/recorders/video_capture.py
Normal file
130
src/recorders/video_capture.py
Normal file
|
|
@ -0,0 +1,130 @@
|
|||
# Top level class for a video capture providing simplified API's for common
|
||||
# functions
|
||||
|
||||
# Import required modules
|
||||
import configparser
|
||||
import cv2
|
||||
import os
|
||||
import sys
|
||||
|
||||
|
||||
# Class to provide boilerplate code to build a video recorder with the
|
||||
# correct settings from the config file.
|
||||
#
|
||||
# The internal recorder can be accessed with 'video_capture.internal'
|
||||
|
||||
|
||||
class VideoCapture:
|
||||
def __init__(self, config):
|
||||
"""
|
||||
Creates a new VideoCapture instance depending on the settings in the
|
||||
provided config file.
|
||||
|
||||
Config can either be a string to the path, or a pre-setup configparser.
|
||||
"""
|
||||
|
||||
# Parse config from string if nedded
|
||||
if isinstance(config, str):
|
||||
self.config = configparser.ConfigParser()
|
||||
self.config.read(config)
|
||||
else:
|
||||
self.config = config
|
||||
|
||||
# Check device path
|
||||
if not os.path.exists(self.config.get("video", "device_path")):
|
||||
print("Camera path is not configured correctly, please edit the 'device_path' config value.")
|
||||
sys.exit(1)
|
||||
|
||||
# Create reader
|
||||
# The internal video recorder
|
||||
self.internal = None
|
||||
# The frame width
|
||||
self.fw = None
|
||||
# The frame height
|
||||
self.fh = None
|
||||
self._create_reader()
|
||||
|
||||
# Request a frame to wake the camera up
|
||||
self.internal.grab()
|
||||
|
||||
def __del__(self):
|
||||
"""
|
||||
Frees resources when destroyed
|
||||
"""
|
||||
if self is not None:
|
||||
self.internal.release()
|
||||
|
||||
def release(self):
|
||||
"""
|
||||
Release cameras
|
||||
"""
|
||||
if self is not None:
|
||||
self.internal.release()
|
||||
|
||||
def read_frame(self):
|
||||
"""
|
||||
Reads a frame, returns the frame and an attempted grayscale conversion of
|
||||
the frame in a tuple:
|
||||
|
||||
(frame, grayscale_frame)
|
||||
|
||||
If the grayscale conversion fails, both items in the tuple are identical.
|
||||
"""
|
||||
|
||||
# Grab a single frame of video
|
||||
# Don't remove ret, it doesn't work without it
|
||||
ret, frame = self.internal.read()
|
||||
if not ret:
|
||||
print("Failed to read camera specified in your 'device_path', aborting")
|
||||
sys.exit(1)
|
||||
|
||||
try:
|
||||
# Convert from color to grayscale
|
||||
# First processing of frame, so frame errors show up here
|
||||
gsframe = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
|
||||
except RuntimeError:
|
||||
gsframe = frame
|
||||
except cv2.error:
|
||||
print("\nAn error occurred in OpenCV\n")
|
||||
raise
|
||||
return frame, gsframe
|
||||
|
||||
def _create_reader(self):
|
||||
"""
|
||||
Sets up the video reader instance
|
||||
"""
|
||||
|
||||
if self.config.get("video", "recording_plugin") == "ffmpeg":
|
||||
# Set the capture source for ffmpeg
|
||||
from recorders.ffmpeg_reader import ffmpeg_reader
|
||||
self.internal = ffmpeg_reader(
|
||||
self.config.get("video", "device_path"),
|
||||
self.config.get("video", "device_format")
|
||||
)
|
||||
|
||||
elif self.config.get("video", "recording_plugin") == "pyv4l2":
|
||||
# Set the capture source for pyv4l2
|
||||
from recorders.pyv4l2_reader import pyv4l2_reader
|
||||
self.internal = pyv4l2_reader(
|
||||
self.config.get("video", "device_path"),
|
||||
self.config.get("video", "device_format")
|
||||
)
|
||||
|
||||
else:
|
||||
# Start video capture on the IR camera through OpenCV
|
||||
self.internal = cv2.VideoCapture(
|
||||
self.config.get("video", "device_path")
|
||||
)
|
||||
|
||||
# Force MJPEG decoding if true
|
||||
if self.config.getboolean("video", "force_mjpeg", fallback=False):
|
||||
# Set a magic number, will enable MJPEG but is badly documentated
|
||||
self.internal.set(cv2.CAP_PROP_FOURCC, 1196444237)
|
||||
|
||||
# Set the frame width and height if requested
|
||||
self.fw = self.config.getint("video", "frame_width", fallback=-1)
|
||||
self.fh = self.config.getint("video", "frame_height", fallback=-1)
|
||||
if self.fw != -1:
|
||||
self.internal.set(cv2.CAP_PROP_FRAME_WIDTH, self.fw)
|
||||
if self.fh != -1:
|
||||
self.internal.set(cv2.CAP_PROP_FRAME_HEIGHT, self.fh)
|
||||
62
src/snapshot.py
Normal file
62
src/snapshot.py
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
# Create and save snapshots of auth attempts
|
||||
|
||||
# Import modules
|
||||
import cv2
|
||||
import os
|
||||
import datetime
|
||||
import numpy as np
|
||||
|
||||
|
||||
def generate(frames, text_lines):
|
||||
"""Generate a shapshot from given frames"""
|
||||
|
||||
# Don't execute if no frames were given
|
||||
if len(frames) == 0:
|
||||
return
|
||||
|
||||
# Get the path to the containing folder
|
||||
abpath = os.path.dirname(os.path.abspath(__file__))
|
||||
# Get frame dimensions
|
||||
frame_height, frame_width, cc = frames[0].shape
|
||||
# Spread the given frames out horizontally
|
||||
snap = np.concatenate(frames, axis=1)
|
||||
|
||||
# Create colors
|
||||
pad_color = [44, 44, 44]
|
||||
text_color = [255, 255, 255]
|
||||
|
||||
# Add a gray square at the bottom of the image
|
||||
snap = cv2.copyMakeBorder(snap, 0, len(text_lines) * 20 + 40, 0, 0, cv2.BORDER_CONSTANT, value=pad_color)
|
||||
|
||||
# Add the Howdy logo if there's space to do so
|
||||
if len(frames) > 1:
|
||||
# Load the logo from file
|
||||
logo = cv2.imread(abpath + "/logo.png")
|
||||
# Calculate the position of the logo
|
||||
logo_y = frame_height + 20
|
||||
logo_x = frame_width * len(frames) - 210
|
||||
|
||||
# Overlay the logo on top of the image
|
||||
snap[logo_y:logo_y+57, logo_x:logo_x+180] = logo
|
||||
|
||||
# Go through each line
|
||||
line_number = 0
|
||||
for line in text_lines:
|
||||
# Calculate how far the line should be from the top
|
||||
padding_top = frame_height + 30 + (line_number * 20)
|
||||
# Print the line onto the image
|
||||
cv2.putText(snap, line, (30, padding_top), cv2.FONT_HERSHEY_SIMPLEX, .4, text_color, 0, cv2.LINE_AA)
|
||||
|
||||
line_number += 1
|
||||
|
||||
# Made sure a snapshot folder exist
|
||||
if not os.path.exists(abpath + "/snapshots"):
|
||||
os.makedirs(abpath + "/snapshots")
|
||||
|
||||
# Generate a filename based on the current time
|
||||
filename = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%S.jpg")
|
||||
# Write the image to that file
|
||||
cv2.imwrite(abpath + "/snapshots/" + filename, snap)
|
||||
|
||||
# Return the saved file location
|
||||
return abpath + "/snapshots/" + filename
|
||||
Loading…
Reference in a new issue