<commit_before>#
#       Univention OpenVPN integration -- openvpn-server.py
#


__package__ = ''  # workaround for PEP 366

import listener
import univention.debug as ud
import re
import univention_baseconfig
import os
import csv
import univention.uldap as ul
from netaddr import *
from univention.config_registry import ConfigRegistry

from datetime import date
from M2Crypto import RSA, BIO
from base64 import b64decode


name        = 'openvpn-server'
description = 'write server-configuration to server.conf and handle address assignment'
filter      = '(objectClass=univentionOpenvpn)'
attributes   = [
    'univentionOpenvpnActive', 'univentionOpenvpnLicense',
    'univentionOpenvpnPort', 'univentionOpenvpnNet', 'univentionOpenvpnNetIPv6',
    'univentionOpenvpnRedirect', 'univentionOpenvpnDuplicate',
    'univentionOpenvpnFixedAddresses', 'univentionOpenvpnUserAddress' ]
modrdn      = 1

action = None

fn_serverconf = '/etc/openvpn/server.conf'


pubbio = BIO.MemoryBuffer('''
-----BEGIN PUBLIC KEY-----
<KEY>
-----END PUBLIC KEY-----
''')
pub = RSA.load_pub_key_bio(pubbio)
pbs = pub.__len__() / 8

def license(key):
  try:
    enc = b64decode(key)
    raw = ''
    while len(enc) > pbs:
      d, key = (enc[:pbs], enc[pbs:])
      raw = raw + pub.public_decrypt(d, 1)
    if len(enc) != pbs:
      return None		# invalid license
    raw = raw + pub.public_decrypt(enc, 1)
    #
    items = raw.rstrip().split('\n')
    if not items:
      return None		# invalid license
    vdate = int(items.pop(0))
    if date.today().toordinal() > vdate:
      ud.debug(ud.LISTENER, ud.ERROR, '3 License has expired')
      return None		# expired
    l = {'valid': True}		# at least one feature returned
    while items:
      kv = items.pop(0).split('=', 1)
      kv.append(True)
      l[kv[0]] = kv[1]
    return l			# valid license
  except:
    return None			# invalid license

def maxvpnusers(key):
  mnlu = 5
  try:
    return max(int(license(key)['u']), mnlu)
  except:
    ud.debug(ud.LISTENER, ud.ERROR, '3 Invalid license')
    return mnlu			# invalid license


# function to open a textfile with setuid(0) for root-action
def load_rc(ofile):
    l = None
    listener.setuid(0)
    try:
        f = open(ofile,"r")
        l = f.readlines()
        f.close()
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to open "%s": %s' % (ofile, str(e)) )
    listener.unsetuid()
    return l

# function to write to a textfile with setuid(0) for root-action
def write_rc(flist, wfile):
    listener.setuid(0)
    try:
        f = open(wfile,"w")
        f.writelines(flist)
        f.close()
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to write to file "%s": %s' % (wfile, str(e)))
    listener.unsetuid()

# function to create a directory with setuid(0) for root-action
def create_dir(path):
    listener.setuid(0)
    try:
        os.makedirs(path)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to make directory "%s": %s' % (path, str(e)))
    listener.unsetuid()

# function to rename a directory with setuid(0) for root-action
def rename_dir(pathold, pathnew):
    listener.setuid(0)
    try:
        os.rename(pathold, pathnew)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to rename directory "%s" to "%s": %s' % (pathold, pathnew, str(e)))
    listener.unsetuid()

# function to delete a textfile with setuid(0) for root-action
def delete_file(fn):
    listener.setuid(0)
    try:
        os.remove(fn)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to remove file "%s": %s' % (fn, str(e)))
    listener.unsetuid()

# function to open an ip map with setuid(0) for root-action
def load_ip_map(path):
    ip_map = []
    listener.setuid(0)
    try:
        with open(path, 'rb') as f:
            r = csv.reader(f, delimiter=' ', quotechar='|')
            for row in r:
                ip_map.append(row)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to load ip map: %s' % str(e))
    listener.unsetuid()
    return ip_map

# function to write an ip map with setuid(0) for root-action
def write_ip_map(ip_map, path):
    listener.setuid(0)
    try:
        with open(path, 'wb') as f:
            w = csv.writer(f, delimiter=' ', quotechar='|', quoting=csv.QUOTE_MINIMAL)
            for i in ip_map:
                w.writerow(i)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to write ip map: %s' % str(e))
    listener.unsetuid()

def handler(dn, new, old, command):
    ud.debug(ud.LISTENER, ud.INFO, '3 server handler')
    global action
    if command == 'n':
        action = None
        return

    if 'univentionOpenvpnActive' in new:
        action = 'restart'
    else:
        action = 'stop'

    cn = new.get('cn', [None])[0]
    myname = listener.baseConfig['hostname']
    if cn != myname:
        action = None
        return

    listener.setuid(0)
    lo = ul.getMachineConnection()

    vpnusers = lo.search('(univentionOpenvpnAccount=1)')
    vpnuc = len(vpnusers)
    maxu = maxvpnusers(new.get('univentionOpenvpnLicense', [None])[0])
    ud.debug(ud.LISTENER, ud.INFO, '3 found %u active openvpn users (%u allowed)' % (vpnuc, maxu))
    if vpnuc > maxu:
        listener.unsetuid()
        action = None
        ud.debug(ud.LISTENER, ud.INFO, '3 skipping actions')
        return			# do nothing


    # activate config
    if not 'univentionOpenvpnActive' in old and os.path.exists(fn_serverconf + '-disabled'):
        listener.setuid(0)
        try:
            os.rename (fn_serverconf + '-disabled', fn_serverconf)
        except Exception, e:
            listener.unsetuid()
            ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to activate server config: %s' % str(e))
            return
        listener.unsetuid()

    if not os.path.exists(fn_serverconf):
        config = """### Constant values

proto udp
dh /etc/openvpn/dh2048.pem
ca /etc/univention/ssl/ucsCA/CAcert.pem
cert /etc/univention/ssl/{hostname}/cert.pem
key /etc/univention/ssl/{hostname}/private.key<commit_msg>fix license handling / allow turning openvpn off<commit_after>#
#       Univention OpenVPN integration -- openvpn-server.py
#


__package__ = ''  # workaround for PEP 366

import listener
import univention.debug as ud
import re
import univention_baseconfig
import os
import csv
import univention.uldap as ul
from netaddr import *
from univention.config_registry import ConfigRegistry

from datetime import date
from M2Crypto import RSA, BIO
from base64 import b64decode


name        = 'openvpn-server'
description = 'write server-configuration to server.conf and handle address assignment'
filter      = '(objectClass=univentionOpenvpn)'
attributes   = [
    'univentionOpenvpnActive', 'univentionOpenvpnLicense',
    'univentionOpenvpnPort', 'univentionOpenvpnNet', 'univentionOpenvpnNetIPv6',
    'univentionOpenvpnRedirect', 'univentionOpenvpnDuplicate',
    'univentionOpenvpnFixedAddresses', 'univentionOpenvpnUserAddress' ]
modrdn      = 1

action = None

fn_serverconf = '/etc/openvpn/server.conf'


pubbio = BIO.MemoryBuffer('''
-----BEGIN PUBLIC KEY-----
<KEY>
-----END PUBLIC KEY-----
''')
pub = RSA.load_pub_key_bio(pubbio)
pbs = pub.__len__() / 8

def license(key):
  try:
    enc = b64decode(key)
    raw = ''
    while len(enc) > pbs:
      d, key = (enc[:pbs], enc[pbs:])
      raw = raw + pub.public_decrypt(d, 1)
    if len(enc) != pbs:
      return None		# invalid license
    raw = raw + pub.public_decrypt(enc, 1)
    #
    items = raw.rstrip().split('\n')
    if not items:
      return None		# invalid license
    vdate = int(items.pop(0))
    if date.today().toordinal() > vdate:
      ud.debug(ud.LISTENER, ud.ERROR, '3 License has expired')
      return None		# expired
    l = {'valid': True}		# at least one feature returned
    while items:
      kv = items.pop(0).split('=', 1)
      kv.append(True)
      l[kv[0]] = kv[1]
    return l			# valid license
  except:
    return None			# invalid license

def maxvpnusers(key):
  mnlu = 5
  try:
    return max(int(license(key)['u']), mnlu)
  except:
    ud.debug(ud.LISTENER, ud.ERROR, '3 Invalid license')
    return mnlu			# invalid license


# function to open a textfile with setuid(0) for root-action
def load_rc(ofile):
    l = None
    listener.setuid(0)
    try:
        f = open(ofile,"r")
        l = f.readlines()
        f.close()
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to open "%s": %s' % (ofile, str(e)) )
    listener.unsetuid()
    return l

# function to write to a textfile with setuid(0) for root-action
def write_rc(flist, wfile):
    listener.setuid(0)
    try:
        f = open(wfile,"w")
        f.writelines(flist)
        f.close()
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to write to file "%s": %s' % (wfile, str(e)))
    listener.unsetuid()

# function to create a directory with setuid(0) for root-action
def create_dir(path):
    listener.setuid(0)
    try:
        os.makedirs(path)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to make directory "%s": %s' % (path, str(e)))
    listener.unsetuid()

# function to rename a directory with setuid(0) for root-action
def rename_dir(pathold, pathnew):
    listener.setuid(0)
    try:
        os.rename(pathold, pathnew)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to rename directory "%s" to "%s": %s' % (pathold, pathnew, str(e)))
    listener.unsetuid()

# function to delete a textfile with setuid(0) for root-action
def delete_file(fn):
    listener.setuid(0)
    try:
        os.remove(fn)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to remove file "%s": %s' % (fn, str(e)))
    listener.unsetuid()

# function to open an ip map with setuid(0) for root-action
def load_ip_map(path):
    ip_map = []
    listener.setuid(0)
    try:
        with open(path, 'rb') as f:
            r = csv.reader(f, delimiter=' ', quotechar='|')
            for row in r:
                ip_map.append(row)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to load ip map: %s' % str(e))
    listener.unsetuid()
    return ip_map

# function to write an ip map with setuid(0) for root-action
def write_ip_map(ip_map, path):
    listener.setuid(0)
    try:
        with open(path, 'wb') as f:
            w = csv.writer(f, delimiter=' ', quotechar='|', quoting=csv.QUOTE_MINIMAL)
            for i in ip_map:
                w.writerow(i)
    except Exception, e:
        ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to write ip map: %s' % str(e))
    listener.unsetuid()

def handler(dn, new, old, command):
    ud.debug(ud.LISTENER, ud.INFO, '3 server handler')
    global action
    if command == 'n':
        action = None
        return

    if 'univentionOpenvpnActive' in new:
        action = 'restart'
    else:
        action = 'stop'

    cn = new.get('cn', [None])[0]
    myname = listener.baseConfig['hostname']
    if cn != myname:
        action = None
        return

    listener.setuid(0)
    lo = ul.getMachineConnection()

    vpnusers = lo.search('(univentionOpenvpnAccount=1)')
    vpnuc = len(vpnusers)
    maxu = maxvpnusers(new.get('univentionOpenvpnLicense', [None])[0])
    ud.debug(ud.LISTENER, ud.INFO, '3 found %u active openvpn users (%u allowed)' % (vpnuc, maxu))
    if vpnuc > maxu:
        listener.unsetuid()
        if action == 'stop':
            ud.debug(ud.LISTENER, ud.INFO, '3 allowing stop action')
        else
            ud.debug(ud.LISTENER, ud.INFO, '3 skipping actions')
            action = None
            return			# do nothing


    # activate config
    if not 'univentionOpenvpnActive' in old and os.path.exists(fn_serverconf + '-disabled'):
        listener.setuid(0)
        try:
            os.rename (fn_serverconf + '-disabled', fn_serverconf)
        except Exception, e:
            listener.unsetuid()
            ud.debug(ud.LISTENER, ud.ERROR, '3 Failed to activate server config: %s' % str(e))
            return
        listener.unsetuid()

    if not os.path.exists(fn_serverconf):
        config = """### Constant values

proto udp
dh /etc/openvpn/dh2048.pem
ca /etc/univention/ssl/ucsCA/CAcert.pem
cert /etc/univention/ssl/{hostname}/cert.pem
key /etc/univention/ssl/{hostname}/private.key
{'fraction_non_alphanumeric': 0.12830645161290322, 'fraction_numerical': 0.009274193548387096, 'mean_word_length': 2.7590178842073354, 'pattern_counts': {'":': 10, '<': 5, '<?xml version=': 0, '>': 11, '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': '1350039', 'n_tokens_mistral': 4389, 'n_tokens_neox': 4253, 'n_words': 1133}