xos/synchronizer/broadbandshield.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import requests
import logging
import json
import sys
from rest_framework.exceptions import APIException

""" format of settings

    ["settings"]
        ["watershed"]
        ["rating"]
        ["categories"]
        ["blocklist"]
        ["allowlist"]

    ["users"]
        array
            ["account_id"] - 58
            ["reporting"] - False
            ["name"] - Scott1
            ["devices"]
            ["settings"] -
                ["watershed"]
                ["rating"]
                ["categories"]
                ["blocklist"]
                ["allowlist"]

    ["devices"]
        array
            ["username"] - "Scott1" or "" if whole-house
            ["uuid"] - empty
            ["mac_address"] - mac address as hex digits in ascii
            ["type"] - "laptop"
            ["name"] - human readable name of device ("Scott's laptop")
            ["settings"]
                 ["watershed"]
                     array
                         array
                             ["rating"]
                             ["category"]
                 ["rating"] - ["G" | "NONE"]
                 ["categories"] - list of categories set by rating
                 ["blocklist"] - []
                 ["allowlist"] - []
"""

class BBS_Failure(APIException):
    status_code=400
    def __init__(self, why="broadbandshield error", fields={}):
        APIException.__init__(self, {"error": "BBS_Failure",
                            "specific_error": why,
                            "fields": fields})


class BBS:
    level_map = {"PG_13": "PG13",
                 "NONE": "OFF",
                 "ALL": "NONE",
                 None: "NONE"}

    def __init__(self, username, password, bbs_hostname=None, bbs_port=None):
        self.username = username
        self.password = password

        # XXX not tested on port 80
        #self.bbs_hostname = "www.broadbandshield.com"
        #self.bbs_port = 80

        if not bbs_hostname:
            bbs_hostname = "cordcompute01.onlab.us"
        if not bbs_port:
            bbs_port = 8018

        self.bbs_hostname = bbs_hostname
        self.bbs_port = int(bbs_port)

        self.api = "http://%s:%d/api" % (self.bbs_hostname, self.bbs_port)
        self.nic_update = "http://%s:%d/nic/update" % (self.bbs_hostname, self.bbs_port)

        self.session = None
        self.settings = None

    def login(self):
        self.session = requests.Session()
        r = self.session.post(self.api + "/login", data = json.dumps({"email": self.username, "password": self.password}))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to login (%d)" % r.status_code)

    def get_account(self):
        if not self.session:
            self.login()

        r = self.session.get(self.api + "/account")
        if (r.status_code != 200):
            raise BBS_Failure("Failed to get account settings (%d)" % r.status_code)
        self.settings = r.json()

        return self.settings

    def post_account(self):
        if not self.settings:
             raise XOSProgrammingError("no settings to post")

        r = self.session.post(self.api + "/account/settings", data= json.dumps(self.settings))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to set account settings (%d)" % r.status_code)

    def add_device(self, name, mac, type="tablet", username=""):
        data = {"name": name, "mac_address": mac, "type": type, "username": username}
        r = self.session.post(self.api + "/device", data = json.dumps(data))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to add device (%d)" % r.status_code)

    def delete_device(self, data):
        r = self.session.delete(self.api + "/device", data = json.dumps(data))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to delete device (%d)" % r.status_code)

    def add_user(self, name, rating="NONE", categories=[]):
        data = {"name": name, "settings": {"rating": rating, "categories": categories}}
        r = self.session.post(self.api + "/users", data = json.dumps(data))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to add user (%d)" % r.status_code)

    def delete_user(self, data):
        r = self.session.delete(self.api + "/users", data = json.dumps(data))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to delete user (%d)" % r.status_code)

    def clear_users_and_devices(self):
        if not self.settings:
            self.get_account()

        for device in self.settings["devices"]:
            self.delete_device(device)

        for user in self.settings["users"]:
            self.delete_user(user)

    def get_whole_home_level(self):
        if not self.settings:
            self.get_account()

        return self.settings["settings"]["rating"]

    def sync(self, whole_home_level, users):
        if not self.settings:
            self.get_account()

        vcpe_users = {}
        for user in users:
            user = user.copy()
            user["level"] = self.level_map.get(user["level"], user["level"])
            user["mac"] = user.get("mac", "")
            vcpe_users[user["name"]] = user

        whole_home_level = self.level_map.get(whole_home_level, whole_home_level)

        if (whole_home_level != self.settings["settings"]["rating"]):
            print "*** set whole_home", whole_home_level, "***"
            self.settings["settings"]["rating"] = whole_home_level
            self.post_account()

        bbs_usernames = [bbs_user["name"] for bbs_user in self.settings["users"]]
        bbs_devicenames = [bbs_device["name"] for bbs_device in self.settings["devices"]]

        add_users = []
        add_devices = []
        delete_users = []
        delete_devices = []

        for bbs_user in self.settings["users"]:
             bbs_username = bbs_user["name"]
             if bbs_username in vcpe_users.keys():
                 vcpe_user = vcpe_users[bbs_username]
                 if bbs_user["settings"]["rating"] != vcpe_user["level"]:
                     print "set user", vcpe_user["name"], "rating", vcpe_user["level"]
                     #bbs_user["settings"]["rating"] = vcpe_user["level"]
                     # add can be used as an update
                     add_users.append(vcpe_user)
             else:
                 delete_users.append(bbs_user)

        for bbs_device in self.settings["devices"]:
             bbs_devicename = bbs_device["name"]
             if bbs_devicename in vcpe_users.keys():
                 vcpe_user = vcpe_users[bbs_devicename]
                 if bbs_device["mac_address"] != vcpe_user["mac"]:
                     print "set device", vcpe_user["name"], "mac", vcpe_user["mac"]
                     #bbs_device["mac_address"] = vcpe_user["mac"]
                     # add of a device can't be used as an update, as you'll end
                     # up with two of them.
                     delete_devices.append(bbs_device)
                     add_devices.append(vcpe_user)
             else:
                 delete_devices.append(bbs_device)

        for (username, user) in vcpe_users.iteritems():
            if not username in bbs_usernames:
                add_users.append(user)
            if not username in bbs_devicenames:
                add_devices.append(user)

        for bbs_user in delete_users:
            print "delete user", bbs_user["name"]
            self.delete_user(bbs_user)

        for bbs_device in delete_devices:
            print "delete device", bbs_device["name"]
            self.delete_device(bbs_device)

        for vcpe_user in add_users:
            print "add user", vcpe_user["name"], "level", vcpe_user["level"]
            self.add_user(vcpe_user["name"], vcpe_user["level"])

        for vcpe_user in add_devices:
            print "add device", vcpe_user["name"], "mac", vcpe_user["mac"]
            self.add_device(vcpe_user["name"], vcpe_user["mac"], "tablet", vcpe_user["name"])

    def get_whole_home_rating(self):
        return self.settings["settings"]["rating"]

    def get_user(self, name):
        for user in self.settings["users"]:
            if user["name"]==name:
                return user
        return None

    def get_device(self, name):
        for device in self.settings["devices"]:
             if device["name"]==name:
                 return device
        return None

    def dump(self):
        if not self.settings:
            self.get_account()

        print "whole_home_rating:", self.settings["settings"]["rating"]
        print "users:"
        for user in self.settings["users"]:
            print "  user", user["name"], "rating", user["settings"]["rating"]

        print "devices:"
        for device in self.settings["devices"]:
            print "  device", device["name"], "user", device["username"], "rating", device["settings"]["rating"], "mac", device["mac_address"]

    def associate(self, ip):
        bbs_hostname = "cordcompute01.onlab.us"
        r = requests.get(self.nic_update, params={"hostname": "onlab.us"}, headers={"X-Forwarded-For": ip}, auth=requests.auth.HTTPBasicAuth(self.username,self.password))
        if (r.status_code != 200):
            raise BBS_Failure("Failed to associate account with ip (%d)" % r.status_code)

def dump():
    bbs = BBS(sys.argv[2], sys.argv[3])
    bbs.dump()

def associate():
    if len(sys.argv)<5:
        print "you need to specify IP address"
        sys.exit(-1)

    bbs = BBS(sys.argv[2], sys.argv[3])
    bbs.associate(sys.argv[4])

def self_test():
    bbs = BBS(sys.argv[2], sys.argv[3])

    print "*** initial ***"
    bbs.dump()

    open("bbs.json","w").write(json.dumps(bbs.settings))

    # a new BBS account will throw a 500 error if it has no rating
    bbs.settings["settings"]["rating"] = "R"
    #bbs.settings["settings"]["category"] = [u'PORNOGRAPHY', u'ADULT', u'ILLEGAL', u'WEAPONS', u'DRUGS', u'GAMBLING', u'CYBERBULLY', u'ANONYMIZERS', u'SUICIDE', u'MALWARE']
    #bbs.settings["settings"]["blocklist"] = []
    #bbs.settings["settings"]["allowlist"] = []
    #for water in bbs.settings["settings"]["watershed"];
    #    water["categories"]=[]
    # delete everything
    bbs.post_account()
    bbs.clear_users_and_devices()

    print "*** cleared ***"
    bbs.settings=None
    bbs.dump()

    users = [{"name": "Moms pc", "level": "R", "mac": "010203040506"},
             {"name": "Dads pc", "level": "R", "mac": "010203040507"},
             {"name": "Jacks ipad", "level": "PG", "mac": "010203040508"},
             {"name": "Jills iphone", "level": "G", "mac": "010203040509"}]

    print "*** syncing mom-R, Dad-R, jack-PG, Jill-G, wholehome-PG-13 ***"

    bbs.settings = None
    bbs.sync("PG-13", users)

    print "*** after sync ***"
    bbs.settings=None
    bbs.dump()
    assert(bbs.get_whole_home_rating() == "PG-13")
    assert(bbs.get_user("Moms pc")["settings"]["rating"] == "R")
    assert(bbs.get_user("Dads pc")["settings"]["rating"] == "R")
    assert(bbs.get_user("Jacks ipad")["settings"]["rating"] == "PG")
    assert(bbs.get_user("Jills iphone")["settings"]["rating"] == "G")
    assert(bbs.get_device("Moms pc")["mac_address"] == "010203040506")
    assert(bbs.get_device("Dads pc")["mac_address"] == "010203040507")
    assert(bbs.get_device("Jacks ipad")["mac_address"] == "010203040508")
    assert(bbs.get_device("Jills iphone")["mac_address"] == "010203040509")

    print "*** update whole home level ***"
    bbs.settings=None
    bbs.get_account()
    bbs.settings["settings"]["rating"] = "PG"
    bbs.post_account()

    print "*** after sync ***"
    bbs.settings=None
    bbs.dump()
    assert(bbs.get_whole_home_rating() == "PG")
    assert(bbs.get_user("Moms pc")["settings"]["rating"] == "R")
    assert(bbs.get_user("Dads pc")["settings"]["rating"] == "R")
    assert(bbs.get_user("Jacks ipad")["settings"]["rating"] == "PG")
    assert(bbs.get_user("Jills iphone")["settings"]["rating"] == "G")
    assert(bbs.get_device("Moms pc")["mac_address"] == "010203040506")
    assert(bbs.get_device("Dads pc")["mac_address"] == "010203040507")
    assert(bbs.get_device("Jacks ipad")["mac_address"] == "010203040508")
    assert(bbs.get_device("Jills iphone")["mac_address"] == "010203040509")

    print "*** delete dad, change moms IP, change jills level to PG, change whole home to PG-13 ***"
    users = [{"name": "Moms pc", "level": "R", "mac": "010203040511"},
             {"name": "Jacks ipad", "level": "PG", "mac": "010203040508"},
             {"name": "Jills iphone", "level": "PG", "mac": "010203040509"}]

    bbs.settings = None
    bbs.sync("PG-13", users)

    print "*** after sync ***"
    bbs.settings=None
    bbs.dump()
    assert(bbs.get_whole_home_rating() == "PG-13")
    assert(bbs.get_user("Moms pc")["settings"]["rating"] == "R")
    assert(bbs.get_user("Dads pc") == None)
    assert(bbs.get_user("Jacks ipad")["settings"]["rating"] == "PG")
    assert(bbs.get_user("Jills iphone")["settings"]["rating"] == "PG")
    assert(bbs.get_device("Moms pc")["mac_address"] == "010203040511")
    assert(bbs.get_device("Dads pc") == None)
    assert(bbs.get_device("Jacks ipad")["mac_address"] == "010203040508")

    print "add dad's laptop"
    users = [{"name": "Moms pc", "level": "R", "mac": "010203040511"},
             {"name": "Dads laptop", "level": "PG-13", "mac": "010203040512"},
             {"name": "Jacks ipad", "level": "PG", "mac": "010203040508"},
             {"name": "Jills iphone", "level": "PG", "mac": "010203040509"}]

    bbs.settings = None
    bbs.sync("PG-13", users)

    print "*** after sync ***"
    bbs.settings=None
    bbs.dump()
    assert(bbs.get_whole_home_rating() == "PG-13")
    assert(bbs.get_user("Moms pc")["settings"]["rating"] == "R")
    assert(bbs.get_user("Dads pc") == None)
    assert(bbs.get_user("Dads laptop")["settings"]["rating"] == "PG-13")
    assert(bbs.get_user("Jacks ipad")["settings"]["rating"] == "PG")
    assert(bbs.get_user("Jills iphone")["settings"]["rating"] == "PG")
    assert(bbs.get_device("Moms pc")["mac_address"] == "010203040511")
    assert(bbs.get_device("Dads pc") == None)
    assert(bbs.get_device("Dads laptop")["mac_address"] == "010203040512")
    assert(bbs.get_device("Jacks ipad")["mac_address"] == "010203040508")

    #bbs.add_user("tom", "G", [u'PORNOGRAPHY', u'ADULT', u'ILLEGAL', u'WEAPONS', u'DRUGS', u'GAMBLING', u'SOCIAL', u'CYBERBULLY', u'GAMES', u'ANONYMIZERS', u'SUICIDE', u'MALWARE'])
    #bbs.add_device(name="tom's iphone", mac="010203040506", type="tablet", username="tom")

def main():
    if len(sys.argv)<4:
        print "syntax: broadbandshield.py <operation> <email> <password>"
        print "        operation = [dump | selftest | assocate"
        sys.exit(-1)

    operation = sys.argv[1]

    if operation=="dump":
        dump()
    elif operation=="selftest":
        self_test()
    elif operation=="associate":
        associate()

if __name__ == "__main__":
    main()