from hmac import compare_digest

from flask import current_app

from flask_httpauth import HTTPBasicAuth, HTTPDigestAuth

basicauth = HTTPBasicAuth()

httpauth = HTTPDigestAuth()

@httpauth.get_password

def get_pw(username: str) -> Optional[str]:

    if username in current_app.config["HTTP_DIGEST_AUTH_USERS"]:

    return None

@basicauth.verify_password

def verify_password(username: str, password: str) -> Optional[bool]:

    if username in current_app.config["HTTP_DIGEST_AUTH_USERS"]:

        return compare_digest(

            current_app.config["HTTP_DIGEST_AUTH_USERS"][username], password

        )

    return None

import json

import jsonschema

from pkg_resources import resource_filename

from .auth import httpauth

from .utils import ActiveStatusv14

sensors_views = Blueprint("sensors", __name__)

ALLOWED_SENSORS_KEYS = json.load(

    open(resource_filename("spaceapi", "schema/sensors.json"), encoding="utf-8")

)

@sensors_views.route("/set/<key>", methods=["POST"])

@httpauth.login_required

    active = ActiveStatusv14()

    if key in ALLOWED_SENSORS_KEYS and key in active["sensors"]:

        try:

            try:

                jsonschema.validate(data, ALLOWED_SENSORS_KEYS[key])

            except jsonschema.ValidationError:

                current_app.logger.error("Validation Error")

                return abort(400)

            if key != "radiation":

                active.set_sensor_value(data, key)

            else:

                active.set_radiation_sensor_value(data)

            active.save_last_state()

            return jsonify(active)

        except ValueError:

            current_app.logger.error("Value Error")

            return abort(400)

    else:

        current_app.logger.error("Sensor not allowed")

        return abort(400)

import email

import email.policy

import json

import os.path

import random

import smtplib

import ssl

from datetime import datetime

from typing import Any, Dict, Iterable, List, Optional

import mastodon

import tweepy

from flask import current_app, request

default_json_file_v14 = os.path.abspath("default_v14.json")

last_state_file_v14 = os.path.abspath("laststate_v14.json")

if not os.path.exists(default_json_file_v14):

    raise RuntimeError("default_v14.json does not exists!")

elif not os.path.isfile(default_json_file_v14):

    raise RuntimeError("default_v14.json is not a file!")

standard_open_message = "The space is now open!"

standard_close_message = "The space is now closed!"

possible_open_messages = (

    standard_open_message,

    "The space is open! Come in and hack something!",

    "Yes, we're open! Come in and create something!",

    "Come by and hack something! We've just opened!",

    "The ChaosStuff is now open for everyone!",

    "Let the Chaos begin! We're open!",

    "What do we hack now? Come and see, we've just opened!",

    "TUWAT! Come and hack. We are open!",

    "It's {time:%H:%M} and we are open! Come and hack!",

    "We just opened our doors at {time:%H:%M}!",

    "Bootup process finished at {time:%H:%M}! Doors opened!",

)

possible_closed_messages = (

    standard_close_message,

    "We're closed now! See you soon.",

    "Sorry, we are closed now!",

    "The ChaosStuff is now closed! Come back another time!",

    "Poweroff process finished! We're closed!",

    "Singularity reached! The space is closed!",

    "Dream of electric sheeps! We are closed!",

    "We closed our doors at {time:%H:%M}. See you soon.",

    "Poweroff process finished at {time:%H:%M}. We're closed!",

)

RADIATON_SUBKEYS = frozenset(("alpha", "beta", "gamma", "beta_gamma"))

class Singleton:

        key = str(hash(cls))

        if not hasattr(cls, "_instance_dict"):

        if key not in cls._instance_dict:

            cls._instance_dict[key] = super().__new__(cls, *args, **kwargs)

        return cls._instance_dict[key]

class ActiveStatusv14(Singleton, dict):

        self.default_json_file = default_json_file_v14

        self.last_state_file = last_state_file_v14

    def reload(self) -> None:

        with open(self.default_json_file, encoding="utf-8") as f:

            self.update(json.load(f))

        if os.path.exists(self.last_state_file) and os.path.isfile(

            self.last_state_file

        ):

            with open(self.last_state_file, encoding="utf-8") as f:

                last_state_json = json.load(f)

            self["state"] = last_state_json["state"]

            self["sensors"].update(last_state_json["sensors"])

    def save_last_state(self) -> None:

        with open(self.last_state_file, mode="w", encoding="utf-8") as f:

            last_state = {}

            last_state["state"] = self["state"]

            last_state["sensors"] = self["sensors"]

            json.dump(last_state, f, sort_keys=True)

    def add_user_present(self, username: str) -> None:

        if self["state"]["open"]:

            if "people_now_present" not in self["sensors"]:

                self["sensors"]["people_now_present"] = [{"value": 0}]

            people_now_present = self["sensors"]["people_now_present"][0]

            if (

                "names" in people_now_present

                and username not in people_now_present["names"]

            ):

                people_now_present["value"] += 1

                if username in current_app.config["PEOPLE_NOW_PRESENT_ALLOWED"]:

                    people_now_present["names"].append(username)

            elif "names" not in people_now_present:

                people_now_present["value"] += 1

                if username in current_app.config["PEOPLE_NOW_PRESENT_ALLOWED"]:

                    people_now_present["names"] = [username]

            self["sensors"]["people_now_present"][0] = people_now_present

        else:

            pass

    def remove_user_present(self, username: str) -> None:

        if self["state"]["open"] and "people_now_present" in self["sensors"]:

            people_now_present = self["sensors"]["people_now_present"][0]

            if people_now_present["value"] > 0:

                people_now_present["value"] -= 1

            if "names" in people_now_present:

                if username in people_now_present["names"]:

                    people_now_present["names"].remove(username)

                if not people_now_present["names"] or people_now_present["value"] == 0:

                    del people_now_present["names"]

            self["sensors"]["people_now_present"][0] = people_now_present

        else:

            pass

    def clear_user_present(self) -> None:

        self["sensors"]["people_now_present"][0]["value"] = 0

        if "names" in self["sensors"]["people_now_present"][0]:

            del self["sensors"]["people_now_present"][0]["names"]

    def notify(self) -> None:

        message = (

            random.choice(possible_open_messages)

            if self["state"]["open"]

            else random.choice(possible_closed_messages)

        )

        message = message.format(time=datetime.now())

        self.send_tweet(message)

        self.send_toot(message)

        subject = (

            standard_open_message if self["state"]["open"] else standard_close_message

        )

        self.send_email(subject, message)

    def send_tweet(self, message: str) -> None:

        if "TWITTER_CONSUMER_KEY" in current_app.config:

            try:

                auth = tweepy.OAuthHandler(

                    current_app.config["TWITTER_CONSUMER_KEY"],

                    current_app.config["TWITTER_CONSUMER_SECRET"],

                )

                auth.set_access_token(

                    current_app.config["TWITTER_ACCESS_TOKEN_KEY"],

                    current_app.config["TWITTER_ACCESS_TOKEN_SECRET"],

                )

                api = tweepy.API(auth)

                api.update_status(

                    message, lat=self["location"]["lat"], long=self["location"]["lon"]

                )

            except Exception as e:

                current_app.logger.error("Sending tweet failed! %s" % e, exc_info=True)

    def send_toot(self, message: str) -> None:

        if "MASTODON_USERCRED_FILE" in current_app.config:

            try:

                api = mastodon.Mastodon(

                    client_id="c3l_spaceapi_clientcred.secret",

                    access_token=current_app.config["MASTODON_USERCRED_FILE"],

                    api_base_url="https://chaos.social",

                )

                api.status_post(message, visibility="unlisted")

            except Exception as e:

                current_app.logger.error("Sending toot failed! %s" % e, exc_info=True)

    def send_email(self, subject: str, message: str) -> None:

        if "EMAIL_PASS" in current_app.config:

            try:

                msg = email.message.EmailMessage(policy=email.policy.default)

                msg["To"] = current_app.config["EMAIL_ANNOUNCE_ADDRESS"]

                email_user = current_app.config["EMAIL_USER"]

                msg["From"] = "spaceapibot <{email}>".format(email=email_user)

                msg["Subject"] = subject

                msg.set_content(message)

                with smtplib.SMTP(

                    current_app.config["EMAIL_HOST"],

                    port=current_app.config["EMAIL_PORT"],

                ) as smtp:

                    smtp.starttls(context=ssl.create_default_context())

                    smtp.login(email_user, current_app.config["EMAIL_PASS"])

                    smtp.send_message(msg)

            except Exception as e:

                current_app.logger.error("Sending email failed! %s" % e, exc_info=True)

    def set_new_state(

        self,

        value: Optional[bool] = None,

        trigger_person: Optional[str] = None,

        message: Optional[str] = None,

    ) -> None:

        if value is not None and value != self["state"]["open"]:

            self["state"]["open"] = value

            self.notify()

            if not value:

                if "people_now_present" in self["sensors"]:

                    self.clear_user_present()

                if "message" in self["state"]:

                    del self["state"]["message"]

            if trigger_person is None:

                if "trigger_person" in self["state"]:

                    del self["state"]["trigger_person"]

            else:

                self["state"]["trigger_person"] = trigger_person

            self["state"]["lastchange"] = int(datetime.now().timestamp())

        if message is not None and message:

            self["state"]["message"] = message

    def set_sensor_value(self, data: Dict[str, Any], key: str) -> None:

        try:

            subkey = first(data, frozenset(("name", "location")))

        except ValueError:

            raise

        try:

            index = fuzzy_list_find(self["sensors"][key], subkey, data[subkey])

            if key == "barometer":

                data["unit"] = "hPa"

            self["sensors"][key][index].update(data)

        except ValueError:

            self["sensors"][key].append(data)

    def set_radiation_sensor_value(self, data: Dict[str, Any]) -> None:

        radiation_keys = [k for k in RADIATON_SUBKEYS if k in data]

        if not radiation_keys:

            raise ValueError

        for first_subkey in radiation_keys: