Files @ e2671a34fdef
Branch filter:

Location: FVDE/ennstatus/ennstatus/api/functions.py

Dennis Fink
More pythonic checking if a value is not in a dictionary

import re
import json

from datetime import datetime
from ast import literal_eval

import pygeoip

FINGERPRINT_REGEX = re.compile(r'^[A-Z0-9]{40}$', re.I)

IP_REGEX = (r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}'
            r'(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$')
IP_REGEX = re.compile(IP_REGEX)

PRIVATE_IP_REGEX = (r'(^127\.)|(^192\.168\.)|(^10\.)|(^172\.1[6-9]\.)'
                    r'|(^172\.2[0-9]\.)|(^172\.3[0-1]\.)|'
                    r'255\.255\.255\.255')
PRIVATE_IP_REGEX = re.compile(PRIVATE_IP_REGEX)

DATE_FORMAT = '%d-%m-%Y %H:%M:%S'

gi4 = pygeoip.GeoIP('/usr/share/GeoIP/GeoIP.dat', pygeoip.MEMORY_CACHE)


def check_json_format(server):

    for key in ('server_type', 'server_name', 'tor_status'):
        if key not in server:
            raise ValueError('%s key not present!\n' % key)

    if server['server_type'] not in ('Exit', 'Relay', 'Bridge'):
        error_message = ('server_type has not the right content!'
                         ' is: %s must be one of: Exit, Relay or Bridge\n') \
            % server['server_type']
        raise ValueError(error_message)

    if not server['tor_status'] in ('Online', 'Offline'):
        error_message = ('tor_status has not the right content!'
                         ' is: %s must be one of: Online or Offline\n') \
            % server['tor_status']

    if server['server_type'] in ('Exit', 'Relay'):
        if 'fingerprint' not in server:
            raise ValueError('fingerprint key not present!\n')
        else:
            if FINGERPRINT_REGEX.match(server['fingerprint']) is None:
                raise ValueError('fingerprint has not the right format!\n')

    else:
        if 'obfs' not in server:
            raise ValueError('obfs key not present!\n')
        else:
            if server['obfs'] not in ('True', 'False'):
                error_message = ('obfs has not the right content!'
                                 ' is: %s must be one of: True or False\n') \
                    % server['obfs']
                raise ValueError(error_message)

    if 'ip' in server:
        if IP_REGEX.match(server['ip']) is None:
            raise ValueError('ip is not the right format!\n')
        elif PRIVATE_IP_REGEX.match(server['ip']) is not None:
            raise ValueError('ip is not accepted!\n')

    return True


def update_server(server, ip):

    server['last_updated'] = datetime.utcnow().strftime(DATE_FORMAT)
    server['server_status'] = 'Online'
    server['country'] = gi4.country_name_by_addr(ip)

    if server['server_type'] == 'Bridge':
        server['obfs'] = literal_eval(server['obfs'])

        for key in ('fingerprint', 'ip'):
            if key in server:
                del server[key]

    else:
        if 'obfs' in server:
            del server['obfs']

        server['ip'] = ip

    try:
        filename = ''.join(['data/', server['server_name'].lower(), '.json'])

        with open(filename, 'w', encoding='utf-8') as fb:
            json.dump(server, fb)

        return server
    except Exception as e:
        return e