Changeset - 2cef966f277b
[Not reviewed]
version_5
0 1 1
Dennis Fink - 10 years ago 2015-08-30 15:32:48
dennis.fink@c3l.lu
Added new check_ip function
2 files changed with 13 insertions and 5 deletions:
0 comments (0 inline, 0 general)
ennstatus/api/model.py
Show inline comments
 
import ipaddress
 
import json
 
import functools
 
import statistics
 

	
 
from pathlib import Path
 
from datetime import datetime
 

	
 
import jsonschema
 
import strict_rfc3339
 
import requests
 

	
 
from flask import current_app
 
from pkg_resources import resource_filename
 

	
 
from ..utils import check_ip
 

	
 

	
 
schema = json.load(
 
    open(
 
        resource_filename('ennstatus.api', 'schema/server.json'),
 
        encoding='utf-8'
 
    )
 
)
 

	
 
validate = functools.partial(
 
    jsonschema.validate,
 
    schema=schema,
 
    format_checker=jsonschema.FormatChecker()
 
)
 

	
 

	
 
def calculate_weight(data):
 

	
 
    obj = {}
 

	
 
    for subkey in ('1_week', '1_month', '3_months', '1_year', '5_years'):
 

	
 
        subdata = data[subkey]
 
        factor = subdata['factor']
 

	
 
        values = [x * factor for x in subdata['values'] if x is not None]
 
@@ -158,54 +161,50 @@ class Server:
 
            if cls.check_json_format(json.loads(server)):
 
                decoded = json.loads(server, cls=ServerDecoder)
 
                return cls(**decoded)
 
        except (jsonschema.ValidationError, ValueError) as e:
 
            raise e
 

	
 
    @classmethod
 
    def from_dict(cls, server):
 
        return cls.from_json(json.dumps(server))
 

	
 
    def json(self):
 
        return json.dumps(self.__dict__, cls=ServerEncoder)
 

	
 
    @staticmethod
 
    def check_json_format(server):
 

	
 
        try:
 
            validate(server)
 
        except jsonschema.ValidationError as e:
 
            raise e
 

	
 
        for key in ('ip', 'ip6'):
 
            if key in server:
 
                address = ipaddress.ip_address(server[key])
 

	
 
                if any({address.is_private, address.is_multicast,
 
                        address.is_unspecified, address.is_reserved,
 
                        address.is_loopback}):
 
                if not check_ip(address):
 
                    raise ValueError('{} is not accepted!\n'.format(key))
 

	
 
        return True
 

	
 
    def save(self):
 

	
 
        filepath = Path('data') / (self.name.lower() + '.json')
 

	
 
        try:
 
            with filepath.open(mode='w', encoding='utf-8') as f:
 
                json.dump(self.__dict__, f, cls=ServerEncoder)
 
        except Exception as e:
 
            raise e
 

	
 
    def update_weights(self):
 

	
 
        if self.type not in ('exit', 'relay'):
 
            raise NotImplementedError
 

	
 
        url = 'https://onionoo.torproject.org/weights?lookup={}'.format(
 
            self.fingerprint
 
        )
 

	
 
        data = requests.get(url)
 

	
 
        try:
ennstatus/utils.py
Show inline comments
 
new file 100644
 

	
 
def check_ip(ipaddress):
 
    return not any({
 
        ipaddress.is_private,
 
        ipaddress.is_multicast,
 
        ipaddress.is_unspecified,
 
        ipaddress.is_reserved,
 
        ipaddress.is_loopback
 
    })
0 comments (0 inline, 0 general)