Coverage for src / local_deep_research / web / server_config.py: 100%

1"""Server configuration management for web app startup. 

2 

3This module handles server configuration that needs to be available before 

4Flask app context is established. It provides a bridge between environment 

5variables and database settings. 

6""" 

7 

8import json 

9import os 

10from pathlib import Path 

11from typing import Dict, Any 

12 

13from loguru import logger 

14 

15from ..config.paths import get_data_dir 

16from ..settings.manager import get_typed_setting_value 

17 

18 

19def get_server_config_path() -> Path: 

20 """Get the path to the server configuration file.""" 

21 return Path(get_data_dir()) / "server_config.json" 

22 

23 

24def load_server_config() -> Dict[str, Any]: 

25 """Load server configuration from file or environment variables. 

26 

27 Returns: 

28 dict: Server configuration with keys: host, port, debug, use_https 

29 """ 

30 config_path = get_server_config_path() 

31 

32 # Try to load from config file first 

33 saved_config = {} 

34 if config_path.exists(): 

35 try: 

36 with open(config_path, "r") as f: 

37 saved_config = json.load(f) 

38 logger.debug(f"Loaded server config from {config_path}") 

39 except Exception as e: 

40 logger.warning(f"Failed to load server config: {e}") 

41 

42 # Ensure correct typing and check environment variables. 

43 config = { 

44 "host": get_typed_setting_value( 

45 "web.host", saved_config.get("host"), "text", default="0.0.0.0" 

46 ), 

47 "port": get_typed_setting_value( 

48 "web.port", saved_config.get("port"), "number", default=5000 

49 ), 

50 "debug": get_typed_setting_value( 

51 "app.debug", saved_config.get("debug"), "checkbox", default=False 

52 ), 

53 "use_https": get_typed_setting_value( 

54 "web.use_https", 

55 saved_config.get("use_https"), 

56 "checkbox", 

57 default=True, 

58 ), 

59 "allow_registrations": get_typed_setting_value( 

60 "app.allow_registrations", 

61 saved_config.get("allow_registrations"), 

62 "checkbox", 

63 default=True, 

64 ), 

65 } 

66 

67 # Security: if LDR_APP_ALLOW_REGISTRATIONS is set to an unrecognized 

68 # value (e.g. "disabled", "nein", typos like "flase"), default to False 

69 # (registrations disabled) rather than True. This "fail closed" approach 

70 # prevents accidental open registration when an admin clearly intended to 

71 # restrict it but used a non-standard boolean string. 

72 _RECOGNIZED_BOOL_VALUES = { 

73 "true", 

74 "false", 

75 "1", 

76 "0", 

77 "yes", 

78 "no", 

79 "on", 

80 "off", 

81 } 

82 raw_reg_env = os.getenv("LDR_APP_ALLOW_REGISTRATIONS") 

83 if raw_reg_env is not None: 

84 normalized = raw_reg_env.lower().strip() 

85 if normalized not in _RECOGNIZED_BOOL_VALUES: 

86 logger.warning( 

87 f"LDR_APP_ALLOW_REGISTRATIONS='{raw_reg_env}' is not a " 

88 f"recognized boolean value. Defaulting to FALSE " 

89 f"(registrations disabled) for security. Use " 

90 f"'true'/'false', '1'/'0', 'yes'/'no', or 'on'/'off'." 

91 ) 

92 config["allow_registrations"] = False 

93 

94 config.update( 

95 { 

96 # Rate limiting settings 

97 "rate_limit_default": get_typed_setting_value( 

98 "security.rate_limit_default", 

99 saved_config.get("rate_limit_default"), 

100 "text", 

101 default="5000 per hour;50000 per day", 

102 ), 

103 "rate_limit_login": get_typed_setting_value( 

104 "security.rate_limit_login", 

105 saved_config.get("rate_limit_login"), 

106 "text", 

107 default="5 per 15 minutes", 

108 ), 

109 "rate_limit_registration": get_typed_setting_value( 

110 "security.rate_limit_registration", 

111 saved_config.get("rate_limit_registration"), 

112 "text", 

113 default="3 per hour", 

114 ), 

115 } 

116 ) 

117 

118 return config 

119 

120 

121def save_server_config(config: Dict[str, Any]) -> None: 

122 """Save server configuration to file. 

123 

124 This should be called when web.host or web.port settings are updated 

125 through the UI. 

126 

127 Args: 

128 config: Server configuration dict 

129 """ 

130 config_path = get_server_config_path() 

131 

132 try: 

133 # Ensure directory exists 

134 config_path.parent.mkdir(parents=True, exist_ok=True) 

135 

136 with open(config_path, "w") as f: 

137 json.dump(config, f, indent=2) 

138 

139 logger.info(f"Saved server config to {config_path}") 

140 except Exception: 

141 logger.exception("Failed to save server config") 

142 

143 

144def sync_from_settings(settings_snapshot: Dict[str, Any]) -> None: 

145 """Sync server config from settings snapshot. 

146 

147 This should be called when settings are updated through the UI. 

148 

149 Args: 

150 settings_snapshot: Settings snapshot containing web.host and web.port 

151 """ 

152 config = load_server_config() 

153 

154 # Update from settings if available 

155 if "web.host" in settings_snapshot: 

156 config["host"] = settings_snapshot["web.host"] 

157 if "web.port" in settings_snapshot: 

158 config["port"] = settings_snapshot["web.port"] 

159 if "app.debug" in settings_snapshot: 

160 config["debug"] = settings_snapshot["app.debug"] 

161 if "web.use_https" in settings_snapshot: 

162 config["use_https"] = settings_snapshot["web.use_https"] 

163 if "app.allow_registrations" in settings_snapshot: 

164 config["allow_registrations"] = settings_snapshot[ 

165 "app.allow_registrations" 

166 ] 

167 if "security.rate_limit_default" in settings_snapshot: 

168 config["rate_limit_default"] = settings_snapshot[ 

169 "security.rate_limit_default" 

170 ] 

171 if "security.rate_limit_login" in settings_snapshot: 

172 config["rate_limit_login"] = settings_snapshot[ 

173 "security.rate_limit_login" 

174 ] 

175 if "security.rate_limit_registration" in settings_snapshot: 

176 config["rate_limit_registration"] = settings_snapshot[ 

177 "security.rate_limit_registration" 

178 ] 

179 

180 save_server_config(config)