<commit_before>#!/usr/bin/env python2
# vim: set noexpandtab:ts=4

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
# Contributors: <EMAIL>

import duo_client
import time
import socket
import sys
import os
import traceback
import syslog
import cPickle as pickle
import ldap

# If FAIL_OPEN is set to True, then authentication will succeed when
# communication with DuoAPI fails.
FAIL_OPEN=True
# User cache allows for caching the authentication validation for USER_CACHE_TIME
# time. This means authentication will succeed and DuoAPI auth() will be bypassed
# for USER_CACHE_TIME after the first successfull authentication.
# Use 0 to disable.
USER_CACHE_TIME=60*60*24
USER_CACHE_PATH="/var/tmp/vpn_user_cache.pickle"
USE_CEF_LOG=True
USERNAME_HACK=True

# Duo settings
IKEY=''
SKEY=''
HOST=''

# LDAP settings
# Auth/login
TRY_LDAP_ONLY_AUTH_FIRST=False
LDAP_URL="ldap://"
LDAP_BIND_DN=''
LDAP_BASE_DN=''

LDAP_CONTROL_BIND_DN=""
LDAP_CONTROL_PASSWORD=""
LDAP_CONTROL_BASE_DN=""
# Only this attr value can log
LDAP_DUOSEC_ATTR_VALUE=""
LDAP_DUOSEC_ATTR=""

def log(msg):
	if USE_CEF_LOG:
		msg = cef(msg)
	syslog.openlog('duo_openvpn', 0, syslog.LOG_DAEMON)
	syslog.syslog(syslog.LOG_INFO, msg)
	syslog.closelog()

def cef(title="DuoAPI", msg="", ext=""):
	hostname = socket.gethostname()
	cefmsg = 'CEF:{v}|{deviceVendor}|{deviceProduct}|{deviceVersion}|{signatureID}|{name}|{message}|{deviceSeverity}|{extension}'.format(
					v='0',
					deviceVendor='Mozilla',
					deviceProduct='OpenVPN',
					deviceVersion='1.0',
					signatureID='0',
					name=title,
					message=msg,
					deviceSeverity='5',
					extension=ext+' dhost=' + hostname,
				)
	return cefmsg


def ldap_attr_get(url, binddn, password, basedn, value_filter, attr):
	conn = ldap.initialize(url)
	try:
		conn.bind_s(binddn, password)
	except ldap.LDAPError, e:
		conn.unbind_s()
		log('LDAP bind failed' % e)
		return None

	try:
		res = conn.search_s(basedn, ldap.SCOPE_SUBTREE, value_filter, [attr])
		#list of attributes
		return res[0][1][attr]
	except:
		log('ldap_get_uid() filter search failed for %s=>%s' % (value_filter, attr))
		return None

def ldap_auth(username, password):
	if (username == None) or (password == None):
		log('User %s LDAP authentication failed' % username)
		return False

	try:
		binddn = LDAP_BIND_DN % username
	except:
		log('Incorrect LDAP_BIND_DN variable')
		return False

	conn = ldap.initialize(LDAP_URL)
	try:
		conn.bind_s(binddn, password)
		conn.unbind_s()
		log('User %s successfully authenticated against LDAP' % username)
		return True
	except ldap.LDAPError:
		conn.unbind_s()
	log('User %s LDAP authentication failed' % username)
	return False

class DuoAPIAuth:
	def __init__(self, ikey, skey, host, username, client_ipaddr, factor, passcode, username_hack, failmode, cache_path, cache_time):
		self.failmode = failmode
		self.passcode = passcode
		if username_hack:
			self.username = self.clean_username(username)
		else:
			self.username = username
		self.client_ipaddr = client_ipaddr
		self.hostname = socket.gethostname()
		self.username = username
		self.client_ipaddr = client_ipaddr
		self.factor = factor
		self.user_cache_time = cache_time

		self.auth_api = duo_client.Auth(
			ikey=ikey,
			skey=skey,
			host=host)

		if os.path.isfile(cache_path):
			self.user_cache = pickle.load(open(cache_path, "rb"))
		else:
			self.user_cache = {}

	def fail_open(self):
		return self.failmode

	def is_auth_cached(self):
		now = time.time()
		if self.user_cache.has_key(self.username):
			try:
				tleft = self.user_cache[self.username]['timestamp']
				ipaddr = self.user_cache[self.username]['ipaddr']
			except:
				return False
			if self.client_ipaddr == ipaddr and tleft > now:
				return True
			del self.user_cache[self.username]
		return False

	def add_auth_cache(self):
		now = time.time()
		self.user_cache[self.username] = {'timestamp': now+self.user_cache_time, 'ipaddr': self.client_ipaddr}
		pickle.dump(self.user_cache, open(USER_CACHE_PATH, "wb"))

	def ping(self):
		now = time.time()
		if not self.auth_api.ping():
			log('DuoAPI not responding')
		end = time.time()-now
		log('DuoAPI responded in %s seconds' % end)

	def check(self):
		if not self.auth_api.check():
			log('DuoAPI IKEY, SKEY or HOST are invalid')

	def clean_username(self, username):
		# use first part of email if an email is present
		try:
			if (username.find('@') != -1):
				username = username.split('@')[0]
		except:
			log('Failed to clean_username()')
			return username
		return username

	def preauth(self):
		res = self.auth_api.preauth(self.username)
		return res['result']

	def doauth(self):
		res = self.auth_api.auth(username=self.username, factor=self.factor, ipaddr=self.client_ipaddr,
							type="OpenVPN login", pushinfo="From%20server="+self.hostname, device="auto", passcode=self.passcode)
		return res

	def auth(self):
		try:
			self.ping()
			self.check()
			auth = self.preauth()
		except socket.error, s:
			log('DuoAPI contact failed %s' % (s))
			return self.fail_open()

		if auth == "allow":
			return True
		elif auth == "enroll":
			log('User %s needs to enroll first' % self.username)
			return False
		elif auth == "auth":
			log('User %s is known - authenticating' % self.username)

			# Auth bypass for cached usernames
			if self.is_auth_cached():<commit_msg>API only accepts "device=" during auth if using push not passcode<commit_after>#!/usr/bin/env python2
# vim: set noexpandtab:ts=4

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
# Contributors: <EMAIL>

import duo_client
import time
import socket
import sys
import os
import traceback
import syslog
import cPickle as pickle
import ldap

# If FAIL_OPEN is set to True, then authentication will succeed when
# communication with DuoAPI fails.
FAIL_OPEN=True
# User cache allows for caching the authentication validation for USER_CACHE_TIME
# time. This means authentication will succeed and DuoAPI auth() will be bypassed
# for USER_CACHE_TIME after the first successfull authentication.
# Use 0 to disable.
USER_CACHE_TIME=60*60*24
USER_CACHE_PATH="/var/tmp/vpn_user_cache.pickle"
USE_CEF_LOG=True
USERNAME_HACK=True

# Duo settings
IKEY=''
SKEY=''
HOST=''

# LDAP settings
# Auth/login
TRY_LDAP_ONLY_AUTH_FIRST=False
LDAP_URL="ldap://"
LDAP_BIND_DN=''
LDAP_BASE_DN=''

LDAP_CONTROL_BIND_DN=""
LDAP_CONTROL_PASSWORD=""
LDAP_CONTROL_BASE_DN=""
# Only this attr value can log
LDAP_DUOSEC_ATTR_VALUE=""
LDAP_DUOSEC_ATTR=""

def log(msg):
	if USE_CEF_LOG:
		msg = cef(msg)
	syslog.openlog('duo_openvpn', 0, syslog.LOG_DAEMON)
	syslog.syslog(syslog.LOG_INFO, msg)
	syslog.closelog()

def cef(title="DuoAPI", msg="", ext=""):
	hostname = socket.gethostname()
	cefmsg = 'CEF:{v}|{deviceVendor}|{deviceProduct}|{deviceVersion}|{signatureID}|{name}|{message}|{deviceSeverity}|{extension}'.format(
					v='0',
					deviceVendor='Mozilla',
					deviceProduct='OpenVPN',
					deviceVersion='1.0',
					signatureID='0',
					name=title,
					message=msg,
					deviceSeverity='5',
					extension=ext+' dhost=' + hostname,
				)
	return cefmsg


def ldap_attr_get(url, binddn, password, basedn, value_filter, attr):
	conn = ldap.initialize(url)
	try:
		conn.bind_s(binddn, password)
	except ldap.LDAPError, e:
		conn.unbind_s()
		log('LDAP bind failed' % e)
		return None

	try:
		res = conn.search_s(basedn, ldap.SCOPE_SUBTREE, value_filter, [attr])
		#list of attributes
		return res[0][1][attr]
	except:
		log('ldap_get_uid() filter search failed for %s=>%s' % (value_filter, attr))
		return None

def ldap_auth(username, password):
	if (username == None) or (password == None):
		log('User %s LDAP authentication failed' % username)
		return False

	try:
		binddn = LDAP_BIND_DN % username
	except:
		log('Incorrect LDAP_BIND_DN variable')
		return False

	conn = ldap.initialize(LDAP_URL)
	try:
		conn.bind_s(binddn, password)
		conn.unbind_s()
		log('User %s successfully authenticated against LDAP' % username)
		return True
	except ldap.LDAPError:
		conn.unbind_s()
	log('User %s LDAP authentication failed' % username)
	return False

class DuoAPIAuth:
	def __init__(self, ikey, skey, host, username, client_ipaddr, factor, passcode, username_hack, failmode, cache_path, cache_time):
		self.failmode = failmode
		self.passcode = passcode
		if username_hack:
			self.username = self.clean_username(username)
		else:
			self.username = username
		self.client_ipaddr = client_ipaddr
		self.hostname = socket.gethostname()
		self.username = username
		self.client_ipaddr = client_ipaddr
		self.factor = factor
		self.user_cache_time = cache_time

		self.auth_api = duo_client.Auth(
			ikey=ikey,
			skey=skey,
			host=host)

		if os.path.isfile(cache_path):
			self.user_cache = pickle.load(open(cache_path, "rb"))
		else:
			self.user_cache = {}

	def fail_open(self):
		return self.failmode

	def is_auth_cached(self):
		now = time.time()
		if self.user_cache.has_key(self.username):
			try:
				tleft = self.user_cache[self.username]['timestamp']
				ipaddr = self.user_cache[self.username]['ipaddr']
			except:
				return False
			if self.client_ipaddr == ipaddr and tleft > now:
				return True
			del self.user_cache[self.username]
		return False

	def add_auth_cache(self):
		now = time.time()
		self.user_cache[self.username] = {'timestamp': now+self.user_cache_time, 'ipaddr': self.client_ipaddr}
		pickle.dump(self.user_cache, open(USER_CACHE_PATH, "wb"))

	def ping(self):
		now = time.time()
		if not self.auth_api.ping():
			log('DuoAPI not responding')
		end = time.time()-now
		log('DuoAPI responded in %s seconds' % end)

	def check(self):
		if not self.auth_api.check():
			log('DuoAPI IKEY, SKEY or HOST are invalid')

	def clean_username(self, username):
		# use first part of email if an email is present
		try:
			if (username.find('@') != -1):
				username = username.split('@')[0]
		except:
			log('Failed to clean_username()')
			return username
		return username

	def preauth(self):
		res = self.auth_api.preauth(self.username)
		return res['result']

	def doauth(self):
		if self.passcode:
			res = self.auth_api.auth(username=self.username, factor=self.factor, ipaddr=self.client_ipaddr,
								type="OpenVPN login", pushinfo="From%20server="+self.hostname, passcode=self.passcode)
		else:
			res = self.auth_api.auth(username=self.username, factor=self.factor, ipaddr=self.client_ipaddr,
								type="OpenVPN login", pushinfo="From%20server="+self.hostname, device="auto")
		return res

	def auth(self):
		try:
			self.ping()
			self.check()
			auth = self.preauth()
		except socket.error, s:
			log('DuoAPI contact failed %s' % (s))
			return self.fail_open()

		if auth == "allow":
			return True
		elif auth == "enroll":
			log('User %s needs to enroll first' % self.username)
			return False
		elif auth == "auth":
			log('User %s is known - authenticating' % self.username)

			# Auth bypass for cached usernames
			if self.is_auth_cached():
{'fraction_non_alphanumeric': 0.12189232695403326, 'fraction_numerical': 0.004650330888928635, 'mean_word_length': 4.487242394504416, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6460074', 'n_tokens_mistral': 4367, 'n_tokens_neox': 3952, 'n_words': 989}