12592571620

Committed 03 Jan 2025 04:04AM UTC coverage: 94.427% (-0.4%) from 94.865%

Build # 12592571620

Build Type

Pull #157

github

Committed by

SeaweedbrainCY

Commit Message

Add build info

Pull Request Pull Request #157: Add healthcheck endpoint

Run Details

18 of 46 new or added lines in 3 files covered. (39.13%)

110 existing lines in 3 files now uncovered.

11081 of 11735 relevant lines covered (94.43%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.0

api/environment.py

import os
import logging
import yaml 
import Utils.env_requirements_check as env_requirements_check
from CryptoClasses.serverRSAKeys import ServerRSAKeys
from Crypto.Protocol.KDF import PBKDF2
from Crypto.Hash import SHA512
import ipaddress
import re

class EnvironmentConfig:
    required_keys = ["type", "config_version", "domain"]
    def __init__(self, data) -> None:
        self.config_version = data["config_version"]
        
        for key in self.required_keys:
            if key not in data:
                logging.error(f"[FATAL] Load config fail. Was expecting the key environment.{key}")
                exit(1)
        self.domain = data["domain"]
        if data["type"] == "local":
            self.type = "local"
            logging.basicConfig(
                format='%(asctime)s %(levelname)-8s %(message)s',
                level=logging.DEBUG,
                datefmt='%Y-%m-%dT%H:%M:%SZ%z')
            logging.debug("Environment set to development")
            if "frontend_URI" not in data:
                logging.error("[FATAL] Load config fail. In local environement, was expecting the key environment.frontend_URI")
                exit(1)
            if "API_URI" not in data:
                logging.error("[FATAL] Load config fail. In local environement, was expecting the key environment.API_URI")
            self.frontend_URI = data["frontend_URI"]
            self.callback_URI = f'{data["API_URI"]}/api/v1/google-drive/oauth/callback'
            os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'
        elif data["type"] == "development":
            self.type = "development"
            logging.basicConfig(
                 filename="/var/log/api/api.log",
                filemode='a',
                format='%(asctime)s %(levelname)-8s %(message)s',
                level=logging.INFO,
                datefmt='%Y-%m-%dT%H:%M:%SZ%z')
            logging.info("Environment set to development")
            self.frontend_URI = f"https://{data['domain']}"
            self.callback_URI = f"https://{data['domain']}/api/v1/google-drive/oauth/callback"
        else:
            self.type = "production"
            logging.basicConfig(
                filename="/var/log/api/api.log",
                filemode='a',
                format='%(asctime)s %(levelname)-8s %(message)s',
                level=logging.INFO,
                datefmt='%Y-%m-%dT%H:%M:%SZ%z')
            self.frontend_URI = f"https://{data['domain']}"
            self.callback_URI = f"https://{data['domain']}/api/v1/google-drive/oauth/callback"


        
class OauthConfig:
    required_keys = ["client_secret_file_path"]

    def __init__(self, data):
        for key in self.required_keys:
            if key not in data:
                logging.error(f"[FATAL] Load config fail. Was expecting the key api.oauth.{key}")
                exit(1)
        self.client_secret_file_path = data["client_secret_file_path"]

class APIConfig:
    required_keys = [ "jwt_secret", "private_key_path", "public_key_path", "flask_secret_key", "server_side_encryption_key"]
    option_config = ["oauth"]

    def __init__(self, data, config_version):
        for key in self.required_keys:
            if key not in data:
                logging.error(f"[FATAL] Load config fail. Was expecting the key api.{key}")
                exit(1)
        for key in self.option_config:
            if key not in data:
                logging.warning(f"api.{key} is not set. Ignoring it ...")
        if "port" not in data:
            logging.warning(f"api.'port' is not set. Using default value: 8080")
            data["port"] = 8080
        
        try:
            self.port = int(data["port"]) 
        except:
            logging.warning("api.port is not valid. Ignoring it. Setting default value: 8080")
            self.port = 8080
        
        self.jwt_secret = data["jwt_secret"]            
        self.private_key_path = data["private_key_path"]
        self.public_key_path = data["public_key_path"]
        self.flask_secret_key = data["flask_secret_key"]
        try:
            if config_version >= 1.0:
                self.server_side_encryption_key = PBKDF2(data["server_side_encryption_key"].encode("utf-8"), '4ATK7mA8aKgT6768' , count=2000000, dkLen=32, hmac_hash_module=SHA512)
            else:
                logging.error(f"[FATAL] Load config fail. config version {config_version} is not supported.")
                exit(1)
        except Exception as e:
            logging.error(f"[FATAL] Load config fail. {e}")
            exit(1)
        if "oauth" in data:
            self.oauth = OauthConfig(data["oauth"])
        else:
            self.oauth = None

        self.trusted_proxy = None
        if "trusted_proxy" in data:
            self.trusted_proxy = []
            for ip in data["trusted_proxy"]:
                try:
                    self.trusted_proxy.append(ipaddress.ip_network(ip))
                except Exception as e:
                    logging.error(f"[FATAL] Load config fail. api.trusted_proxy contains an invalid ip address. {e}")
                    exit(1)
        self.session_token_validity = 600
        if "session_token_validity" in data:
            try:
                self.session_token_validity = int(data["session_token_validity"])
            except Exception as e:
                logging.error(f"[FATAL] Load config fail. api.session_token_validity is not valid. {e}")
                exit(1)
        
        self.refresh_token_validity = 86400
        if "refresh_token_validity" in data:
            try:
                self.refresh_token_validity = int(data["refresh_token_validity"])
            except Exception as e:
                logging.error(f"[FATAL] Load config fail. api.refresh_token_validity is not valid. {e}")
                exit(1)
        
        if "health_check" in data:
            if 'node_check_enabled':
                self.node_check_enabled = data["health_check"]["node_check_enabled"]
                if self.node_check_enabled:
                    required_node_health_check_keys = ["node_name", "node_name_hmac_secret"]
                    for key in required_node_health_check_keys:
                        if key not in data["health_check"]:
                            logging.error(f"[FATAL] Load config fail. api.health_check.node_check_enabled is True so api.health_check require the key {key} to exist.")
                            exit(1)
                    self.node_name = data["health_check"]["node_name"]
                    self.node_name_hmac_secret = data["health_check"]["node_name_hmac_secret"]
            else:
                logging.error(f"[FATAL] Load config fail. api.health_check require the key node_check_enabled to exist. {e}")
                exit(1)
        else: 
            self.node_check_enabled = False

        self.version = "0.0.0"
        self.build = "0000000"
        with open("VERSION", "r") as f:
            version = f.readline().strip()
            if re.match(r"^b?v?\d+\.\d+\.\d+(-[a-zA-Z0-9]+)?$", version) or version == "dev":
                self.version = version
            else:
                logging.warning(f"VERSION file is not in the correct format. Using default value: 0.0.0")
            build = f.readline().strip() # github short sha1
            if re.match(r"^[a-f0-9]{7}$", build):
                self.build = build
            else:
                logging.warning(f"VERSION file is not in the correct format. Using default build: 0000000")

                
        
class DatabaseConfig:
    required_keys = ["database_uri"]
    def __init__(self, data):
        for key in self.required_keys:
            if key not in data:
                logging.error(f"[FATAL] Load config fail. Was expecting the key database.{key}")
                exit(1)
        self.database_uri = data["database_uri"]
        self.are_all_tables_created = False



class EmailsConfig:
    required_keys = ["email_sender_address", "email_smtp_password", "email_smtp_server", "email_smtp_port", "email_smtp_username"]
    def __init__(self, data):
        if "require_email_validation" not in data:
            data["require_email_validation"] = False
        if data["require_email_validation"] == False:
            logging.warning("require_email_validation is disabled. Users will not be asked to verify their email address. You can enable this option by setting the require_email_validation  variable to true at any moment.")
            self.require_email_validation = False
        else :
            self.require_email_validation = True
            for key in self.required_keys:
                if key not in data:
                    logging.error(f"[FATAL] Load config fail. Was expecting the key features.emails.{key} because require_email_validation is set to true")
                    exit(1)
            self.sender_address = data["email_sender_address"]
            self.sender_password = data["email_smtp_password"]
            self.smtp_server = data["email_smtp_server"]
            self.smtp_port = data["email_smtp_port"]
            self.smtp_username = data["email_smtp_username"]

class RateLimitingConfig:
    def __init__(self, data):
        try :
            self.login_attempts_limit_per_ip = int(data["login_attempts_limit_per_ip"]) if "login_attempts_limit_per_ip" in data else 10
            self.send_email_attempts_limit_per_user = int(data["send_email_attempts_limit_per_user"]) if "send_email_attempts_limit_per_user" in data else 5
            self.login_ban_time = int(data["login_ban_time"]) if "login_ban_time" in data else 15
            self.email_ban_time = int(data["email_ban_time"]) if "email_ban_time" in data else 60
        except Exception as e:
            logging.error(f"[FATAL] Load config fail. {e}")
            exit(1)


class SentryConfig:
    required_keys = ["dsn"]
    def __init__(self, data):
        for key in self.required_keys:
            if key not in data:
                logging.error(f"[FATAL] Load config fail. Was expecting the key features.sentry.{key}")
                exit(1)
        self.dsn = data["dsn"]


class FeaturesConfig:
    def __init__(self, data):
                
        self.emails = EmailsConfig(data["emails"]) if "emails" in data else None
        self.rate_limiting = RateLimitingConfig(data["rate_limiting"] if "rate_limiting" in data else [])
        self.sentry = SentryConfig(data["sentry"]) if "sentry" in data else None


class Config:
    required_keys = ["api", "environment", "database", "features"]
    def __init__(self, data):
        for key in self.required_keys:
            if key not in data:
                logging.error(f"[FATAL] Load config fail. Was expecting the key {key}")
                exit(1)
        self.environment = EnvironmentConfig(data["environment"] if data["environment"] != None else [])
        self.api = APIConfig(data["api"] if data["api"] != None else [], self.environment.config_version)
        self.database = DatabaseConfig(data["database"] if data["database"] != None else [])
        self.features = FeaturesConfig(data["features"] if data["features"] != None else [])




try:
    with open("./config/config.yml") as config_yml:
        try:
            raw_conf = yaml.safe_load(config_yml)
            conf = Config(raw_conf)
        
        except yaml.YAMLError as exc:
            raise Exception(exc)
except Exception as e :
    logging.error(f"[FATAL] API will stop now. Error while checking /api/config/config.yml, {e}")
    exit(1)


env_requirements_check.test_conf(conf) 

1	import os	1✔
2	import logging	1✔
3	import yaml	1✔
4	import Utils.env_requirements_check as env_requirements_check	1✔
5	from CryptoClasses.serverRSAKeys import ServerRSAKeys	1✔
6	from Crypto.Protocol.KDF import PBKDF2	1✔
7	from Crypto.Hash import SHA512	1✔
8	import ipaddress	1✔
9	import re	1✔
10
11	class EnvironmentConfig:	1✔
12	required_keys = ["type", "config_version", "domain"]	1✔
13	def __init__(self, data) -> None:	1✔
14	self.config_version = data["config_version"]	1✔
15
16	for key in self.required_keys:	1✔
17	if key not in data:	1✔
18	logging.error(f"[FATAL] Load config fail. Was expecting the key environment.{key}")	×
UNCOV 19	exit(1)	×
20	self.domain = data["domain"]	1✔
21	if data["type"] == "local":	1✔
22	self.type = "local"	×
UNCOV 23	logging.basicConfig(	×
24	format='%(asctime)s %(levelname)-8s %(message)s',
25	level=logging.DEBUG,
26	datefmt='%Y-%m-%dT%H:%M:%SZ%z')
27	logging.debug("Environment set to development")	×
28	if "frontend_URI" not in data:	×
29	logging.error("[FATAL] Load config fail. In local environement, was expecting the key environment.frontend_URI")	×
30	exit(1)	×
31	if "API_URI" not in data:	×
32	logging.error("[FATAL] Load config fail. In local environement, was expecting the key environment.API_URI")	×
33	self.frontend_URI = data["frontend_URI"]	×
34	self.callback_URI = f'{data["API_URI"]}/api/v1/google-drive/oauth/callback'	×
UNCOV 35	os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'	×
36	elif data["type"] == "development":	1✔
37	self.type = "development"	1✔
38	logging.basicConfig(	1✔
39	filename="/var/log/api/api.log",
40	filemode='a',
41	format='%(asctime)s %(levelname)-8s %(message)s',
42	level=logging.INFO,
43	datefmt='%Y-%m-%dT%H:%M:%SZ%z')
44	logging.info("Environment set to development")	1✔
45	self.frontend_URI = f"https://{data['domain']}"	1✔
46	self.callback_URI = f"https://{data['domain']}/api/v1/google-drive/oauth/callback"	1✔
47	else:
48	self.type = "production"	×
UNCOV 49	logging.basicConfig(	×
50	filename="/var/log/api/api.log",
51	filemode='a',
52	format='%(asctime)s %(levelname)-8s %(message)s',
53	level=logging.INFO,
54	datefmt='%Y-%m-%dT%H:%M:%SZ%z')
55	self.frontend_URI = f"https://{data['domain']}"	×
UNCOV 56	self.callback_URI = f"https://{data['domain']}/api/v1/google-drive/oauth/callback"	×
57
58
59
60	class OauthConfig:	1✔
61	required_keys = ["client_secret_file_path"]	1✔
62
63	def __init__(self, data):	1✔
64	for key in self.required_keys:	1✔
65	if key not in data:	1✔
66	logging.error(f"[FATAL] Load config fail. Was expecting the key api.oauth.{key}")	×
UNCOV 67	exit(1)	×
68	self.client_secret_file_path = data["client_secret_file_path"]	1✔
69
70	class APIConfig:	1✔
71	required_keys = [ "jwt_secret", "private_key_path", "public_key_path", "flask_secret_key", "server_side_encryption_key"]	1✔
72	option_config = ["oauth"]	1✔
73
74	def __init__(self, data, config_version):	1✔
75	for key in self.required_keys:	1✔
76	if key not in data:	1✔
77	logging.error(f"[FATAL] Load config fail. Was expecting the key api.{key}")	×
UNCOV 78	exit(1)	×
79	for key in self.option_config:	1✔
80	if key not in data:	1✔
UNCOV 81	logging.warning(f"api.{key} is not set. Ignoring it ...")	×
82	if "port" not in data:	1✔
83	logging.warning(f"api.'port' is not set. Using default value: 8080")	1✔
84	data["port"] = 8080	1✔
85
86	try:	1✔
87	self.port = int(data["port"])	1✔
88	except:	×
89	logging.warning("api.port is not valid. Ignoring it. Setting default value: 8080")	×
UNCOV 90	self.port = 8080	×
91
92	self.jwt_secret = data["jwt_secret"]	1✔
93	self.private_key_path = data["private_key_path"]	1✔
94	self.public_key_path = data["public_key_path"]	1✔
95	self.flask_secret_key = data["flask_secret_key"]	1✔
96	try:	1✔
97	if config_version >= 1.0:	1✔
98	self.server_side_encryption_key = PBKDF2(data["server_side_encryption_key"].encode("utf-8"), '4ATK7mA8aKgT6768' , count=2000000, dkLen=32, hmac_hash_module=SHA512)	1✔
99	else:
100	logging.error(f"[FATAL] Load config fail. config version {config_version} is not supported.")	×
101	exit(1)	×
102	except Exception as e:	×
103	logging.error(f"[FATAL] Load config fail. {e}")	×
UNCOV 104	exit(1)	×
105	if "oauth" in data:	1✔
106	self.oauth = OauthConfig(data["oauth"])	1✔
107	else:
UNCOV 108	self.oauth = None	×
109
110	self.trusted_proxy = None	1✔
111	if "trusted_proxy" in data:	1✔
112	self.trusted_proxy = []	×
113	for ip in data["trusted_proxy"]:	×
114	try:	×
115	self.trusted_proxy.append(ipaddress.ip_network(ip))	×
116	except Exception as e:	×
117	logging.error(f"[FATAL] Load config fail. api.trusted_proxy contains an invalid ip address. {e}")	×
UNCOV 118	exit(1)	×
119	self.session_token_validity = 600	1✔
120	if "session_token_validity" in data:	1✔
121	try:	×
122	self.session_token_validity = int(data["session_token_validity"])	×
123	except Exception as e:	×
124	logging.error(f"[FATAL] Load config fail. api.session_token_validity is not valid. {e}")	×
UNCOV 125	exit(1)	×
126
127	self.refresh_token_validity = 86400	1✔
128	if "refresh_token_validity" in data:	1✔
129	try:	×
130	self.refresh_token_validity = int(data["refresh_token_validity"])	×
131	except Exception as e:	×
132	logging.error(f"[FATAL] Load config fail. api.refresh_token_validity is not valid. {e}")	×
UNCOV 133	exit(1)	×
134
135	if "health_check" in data:	1✔
UNCOV 136	if 'node_check_enabled':	×
UNCOV 137	self.node_check_enabled = data["health_check"]["node_check_enabled"]	×
UNCOV 138	if self.node_check_enabled:	×
UNCOV 139	required_node_health_check_keys = ["node_name", "node_name_hmac_secret"]	×
140	for key in required_node_health_check_keys:	×
141	if key not in data["health_check"]:	×
UNCOV 142	logging.error(f"[FATAL] Load config fail. api.health_check.node_check_enabled is True so api.health_check require the key {key} to exist.")	×
UNCOV 143	exit(1)	×
UNCOV 144	self.node_name = data["health_check"]["node_name"]	×
UNCOV 145	self.node_name_hmac_secret = data["health_check"]["node_name_hmac_secret"]	×
146	else:
147	logging.error(f"[FATAL] Load config fail. api.health_check require the key node_check_enabled to exist. {e}")
148	exit(1)
149	else:
150	self.node_check_enabled = False	1✔
151
152	self.version = "0.0.0"	1✔
153	self.build = "0000000"	1✔
154	with open("VERSION", "r") as f:	1✔
155	version = f.readline().strip()	1✔
156	if re.match(r"^b?v?\d+\.\d+\.\d+(-[a-zA-Z0-9]+)?$", version) or version == "dev":	1✔
157	self.version = version	1✔
158	else:
159	logging.warning(f"VERSION file is not in the correct format. Using default value: 0.0.0")	×
160	build = f.readline().strip() # github short sha1	1✔
161	if re.match(r"^[a-f0-9]{7}$", build):	1✔
162	self.build = build	1✔
163	else:
UNCOV 164	logging.warning(f"VERSION file is not in the correct format. Using default build: 0000000")	×
165
166
167
168	class DatabaseConfig:	1✔
169	required_keys = ["database_uri"]	1✔
170	def __init__(self, data):	1✔
171	for key in self.required_keys:	1✔
172	if key not in data:	1✔
UNCOV 173	logging.error(f"[FATAL] Load config fail. Was expecting the key database.{key}")	×
174	exit(1)	×
175	self.database_uri = data["database_uri"]	1✔
176	self.are_all_tables_created = False	1✔
177
178
179
180	class EmailsConfig:	1✔
181	required_keys = ["email_sender_address", "email_smtp_password", "email_smtp_server", "email_smtp_port", "email_smtp_username"]	1✔
182	def __init__(self, data):	1✔
183	if "require_email_validation" not in data:	1✔
184	data["require_email_validation"] = False	×
185	if data["require_email_validation"] == False:	1✔
186	logging.warning("require_email_validation is disabled. Users will not be asked to verify their email address. You can enable this option by setting the require_email_validation variable to true at any moment.")	×
UNCOV 187	self.require_email_validation = False	×
188	else :
189	self.require_email_validation = True	1✔
190	for key in self.required_keys:	1✔
191	if key not in data:	1✔
UNCOV 192	logging.error(f"[FATAL] Load config fail. Was expecting the key features.emails.{key} because require_email_validation is set to true")	×
UNCOV 193	exit(1)	×
194	self.sender_address = data["email_sender_address"]	1✔
195	self.sender_password = data["email_smtp_password"]	1✔
196	self.smtp_server = data["email_smtp_server"]	1✔
197	self.smtp_port = data["email_smtp_port"]	1✔
198	self.smtp_username = data["email_smtp_username"]	1✔
199
200	class RateLimitingConfig:	1✔
201	def __init__(self, data):	1✔
202	try :	1✔
203	self.login_attempts_limit_per_ip = int(data["login_attempts_limit_per_ip"]) if "login_attempts_limit_per_ip" in data else 10	1✔
204	self.send_email_attempts_limit_per_user = int(data["send_email_attempts_limit_per_user"]) if "send_email_attempts_limit_per_user" in data else 5	1✔
205	self.login_ban_time = int(data["login_ban_time"]) if "login_ban_time" in data else 15	1✔
206	self.email_ban_time = int(data["email_ban_time"]) if "email_ban_time" in data else 60	1✔
UNCOV 207	except Exception as e:	×
UNCOV 208	logging.error(f"[FATAL] Load config fail. {e}")	×
UNCOV 209	exit(1)	×
210
211
212	class SentryConfig:	1✔
213	required_keys = ["dsn"]	1✔
214	def __init__(self, data):	1✔
UNCOV 215	for key in self.required_keys:	×
UNCOV 216	if key not in data:	×
UNCOV 217	logging.error(f"[FATAL] Load config fail. Was expecting the key features.sentry.{key}")	×
218	exit(1)	×
219	self.dsn = data["dsn"]	×
220
221
222	class FeaturesConfig:	1✔
223	def __init__(self, data):	1✔
224
225	self.emails = EmailsConfig(data["emails"]) if "emails" in data else None	1✔
226	self.rate_limiting = RateLimitingConfig(data["rate_limiting"] if "rate_limiting" in data else [])	1✔
227	self.sentry = SentryConfig(data["sentry"]) if "sentry" in data else None	1✔
228
229
230	class Config:	1✔
231	required_keys = ["api", "environment", "database", "features"]	1✔
232	def __init__(self, data):	1✔
233	for key in self.required_keys:	1✔
234	if key not in data:	1✔
UNCOV 235	logging.error(f"[FATAL] Load config fail. Was expecting the key {key}")	×
UNCOV 236	exit(1)	×
237	self.environment = EnvironmentConfig(data["environment"] if data["environment"] != None else [])	1✔
238	self.api = APIConfig(data["api"] if data["api"] != None else [], self.environment.config_version)	1✔
239	self.database = DatabaseConfig(data["database"] if data["database"] != None else [])	1✔
240	self.features = FeaturesConfig(data["features"] if data["features"] != None else [])	1✔
241
242
243
244
245	try:	1✔
246	with open("./config/config.yml") as config_yml:	1✔
247	try:	1✔
248	raw_conf = yaml.safe_load(config_yml)	1✔
249	conf = Config(raw_conf)	1✔
250
UNCOV 251	except yaml.YAMLError as exc:	×
UNCOV 252	raise Exception(exc)	×
UNCOV 253	except Exception as e :	×
UNCOV 254	logging.error(f"[FATAL] API will stop now. Error while checking /api/config/config.yml, {e}")	×
UNCOV 255	exit(1)	×
256
257
258	env_requirements_check.test_conf(conf)	1✔

SeaweedbrainCY / zero-totp / 12592571620

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous