Coverage for src/lektor_ng/admin/modules/serve.py: 100%

1from __future__ import annotations 

2 

3import dataclasses 

4import mimetypes 

5import os 

6import re 

7from pathlib import Path 

8from typing import TYPE_CHECKING 

9from zlib import adler32 

10 

11from flask import ( 

12 Blueprint, 

13 Response, 

14 abort, 

15 current_app, 

16 render_template, 

17 request, 

18 send_file, 

19 url_for, 

20) 

21from werkzeug.exceptions import NotFound 

22from werkzeug.security import safe_join 

23from werkzeug.utils import append_slash_redirect 

24 

25from lektor_ng.admin.context import LektorApp, LektorContext, get_lektor_context 

26from lektor_ng.assets import Asset, Directory 

27from lektor_ng.constants import PRIMARY_ALT 

28from lektor_ng.db import Record 

29 

30if TYPE_CHECKING: 

31 from flask.typing import ResponseReturnValue, ResponseValue 

32 

33 from lektor_ng.builder import Artifact 

34 from lektor_ng.buildfailures import BuildFailure 

35 from lektor_ng.sourceobj import SourceObject 

36 

37 

38bp = Blueprint("serve", __name__) 

39 

40 

41Filename = str | os.PathLike[str] 

42 

43 

44@dataclasses.dataclass(frozen=True) 

45class LivereloadConfig: 

46 artifactName: str 

47 eventsUrl: str 

48 workerJs: str 

49 

50 @classmethod 

51 def from_artifact(cls, artifact: Artifact | None) -> LivereloadConfig | None: 

52 if artifact is not None and "livereload" in current_app.blueprints: 

53 return cls( 

54 artifactName=artifact.artifact_name, 

55 eventsUrl=url_for("livereload.events"), 

56 workerJs=url_for("static", filename="livereload-worker.js"), 

57 ) 

58 return None 

59 

60 

61@dataclasses.dataclass(frozen=True) 

62class TooldrawerConfig: 

63 editUrl: str | None = None 

64 livereloadConfig: LivereloadConfig | None = None 

65 

66 def __bool__(self) -> bool: 

67 return self.editUrl is not None or self.livereloadConfig is not None 

68 

69 

70def _inject_tooldrawer(html: bytes, tooldrawer_config: TooldrawerConfig | None) -> bytes: 

71 """Add "edit pencil" and "livereload" control buttons to the text of an HTML 

72 page.""" 

73 if tooldrawer_config: 

74 tooldrawer_html = render_template( 

75 "tooldrawer.html", 

76 tooldrawer_config=dataclasses.asdict(tooldrawer_config), 

77 tooldrawer_js=url_for("static", filename="tooldrawer.js"), 

78 ).encode("utf-8") 

79 match = re.search(rb"(?i)</\s*head\s*>|\Z", html) 

80 assert match is not None 

81 head_end = match.start() 

82 html = html[:head_end] + tooldrawer_html + html[head_end:] 

83 return html 

84 

85 

86def _send_html_for_editing( 

87 artifact: Artifact, tooldrawer_config: TooldrawerConfig, mimetype: str = "text/html" 

88) -> ResponseValue: 

89 """Serve an HTML file, after mangling it to add an "edit pencil" button.""" 

90 try: 

91 with open(artifact.dst_filename, "rb") as fp: 

92 html = fp.read() 

93 st = os.stat(fp.fileno()) 

94 except (FileNotFoundError, IsADirectoryError, PermissionError): 

95 abort(404) 

96 html = _inject_tooldrawer(html, tooldrawer_config) 

97 check = adler32(f"{artifact.dst_filename}\0{hash(tooldrawer_config)}".encode()) & 0xFFFFFFFF 

98 resp = Response(html, mimetype=mimetype) 

99 resp.set_etag(f"{st.st_mtime}-{st.st_size}-{check}") 

100 return resp 

101 

102 

103def _deduce_mimetype(filename: Filename) -> str: 

104 mimetype = mimetypes.guess_type(filename)[0] 

105 if mimetype is None: 

106 mimetype = "application/octet-stream" 

107 return mimetype 

108 

109 

110def _checked_send_file(filename: Filename, mimetype: str | None = None) -> ResponseValue: 

111 """Same as flask.send_file, except raises NotFound on file errors.""" 

112 # NB: flask.send_file interprets relative paths relative to 

113 # current_app.root_path. We don't want that. 

114 try: 

115 resp = send_file(os.path.abspath(filename), mimetype=mimetype) 

116 except (FileNotFoundError, IsADirectoryError, PermissionError): 

117 abort(404) 

118 return resp 

119 

120 

121class HiddenRecordException(NotFound): 

122 """Exception thrown when a request is made for a hidden page.""" 

123 

124 def __init__(self, source: SourceObject) -> None: 

125 super().__init__(description=f"Record is hidden: {source!r}") 

126 self.source = source 

127 

128 

129class ArtifactServer: 

130 """Resolve url_path to a Lektor source object, build it, serve the result. 

131 

132 Redirects to slash-appended path if appropriate. 

133 

134 Raises NotFound if source object can not be resolved, or if it does not 

135 produce an artifact. 

136 

137 """ 

138 

139 def __init__(self, lektor_context: LektorContext) -> None: 

140 self.lektor_ctx = lektor_context 

141 

142 def resolve_url_path(self, url_path: str) -> SourceObject: 

143 """Resolve URL path to a source object. 

144 

145 Raise NotFound if resolution fails. 

146 """ 

147 source = self.lektor_ctx.pad.resolve_url_path(url_path, include_invisible=True) 

148 if source is None: 

149 abort(404) 

150 return source 

151 

152 @staticmethod 

153 def resolve_directory_index(directory: Directory) -> Asset: 

154 """Find an index.html (or equivalent) asset for a Directory asset 

155 

156 Raise NotFound if no index is found. 

157 """ 

158 for name in "index.html", "index.htm": 

159 index = directory.resolve_url_path([name]) 

160 if index is not None: 

161 break 

162 else: 

163 abort(404) 

164 return index 

165 

166 def build_primary_artifact(self, source: SourceObject) -> tuple[Artifact, BuildFailure | None]: 

167 """Build source object, return primary artifact. 

168 

169 If the build was successfull, returns a tuple of (artifact, ``None``). 

170 

171 If the build failed, returns a tuple of (artifact, failure), 

172 where failure is an instance of ``BuildFailure`` which 

173 contains information regarding the failure. 

174 

175 Raises NotFound if no primary artifact is produced by the build process. 

176 """ 

177 lektor_ctx = self.lektor_ctx 

178 with lektor_ctx.cli_reporter(): 

179 prog, _ = lektor_ctx.builder.build(source) 

180 artifact = prog.primary_artifact 

181 if artifact is None: 

182 abort(404) 

183 failure = lektor_ctx.failure_controller.lookup_failure(artifact.artifact_name) 

184 return artifact, failure 

185 

186 @staticmethod 

187 def handle_build_failure(failure: BuildFailure, tooldrawer_config: TooldrawerConfig | None = None) -> Response: 

188 """Format build failure to an HTML response.""" 

189 html = render_template("build-failure.html", **failure.data).encode("utf-8") 

190 html = _inject_tooldrawer(html, tooldrawer_config) 

191 return Response(html, mimetype="text/html") 

192 

193 def get_edit_url(self, source: SourceObject) -> str | None: 

194 primary_alternative = self.lektor_ctx.config.primary_alternative 

195 if not isinstance(source, Record): 

196 # Asset or VirtualSourceObject — not editable 

197 return None 

198 record = source.record 

199 alt = record.alt if record.alt not in (PRIMARY_ALT, primary_alternative) else None 

200 return url_for("url.edit", path=record.path, alt=alt) 

201 

202 def serve_artifact(self, url_path: str) -> ResponseValue: 

203 source = self.resolve_url_path(url_path) 

204 

205 # If the request path does not end with a slash but we 

206 # requested a URL that actually wants a trailing slash, we 

207 # append it. This is consistent with what apache and nginx do 

208 # and it ensures our relative urls work. 

209 if not url_path.endswith("/") and source.url_path.endswith("/") and source.url_path != "/": 

210 return append_slash_redirect(request.environ) 

211 

212 if source.is_hidden: 

213 raise HiddenRecordException(source) 

214 

215 if isinstance(source, Directory): 

216 # Special case for asset directories: resolve to index.html 

217 source = self.resolve_directory_index(source) 

218 

219 artifact, failure = self.build_primary_artifact(source) 

220 tooldrawer_config = TooldrawerConfig( 

221 editUrl=self.get_edit_url(source), 

222 livereloadConfig=LivereloadConfig.from_artifact(artifact), 

223 ) 

224 

225 # If there was a build failure for the given artifact, we want 

226 # to render this instead of sending the (most likely missing or 

227 # corrupted) file. 

228 if failure is not None: 

229 return self.handle_build_failure(failure, tooldrawer_config) 

230 

231 mimetype = _deduce_mimetype(artifact.dst_filename) 

232 if mimetype == "text/html" and tooldrawer_config: 

233 return _send_html_for_editing(artifact, tooldrawer_config, mimetype) 

234 return _checked_send_file(artifact.dst_filename, mimetype=mimetype) 

235 

236 

237def serve_artifact(path: str) -> ResponseValue: 

238 lektor_context = get_lektor_context() 

239 return ArtifactServer(lektor_context).serve_artifact(path) 

240 

241 

242def serve_file(path: str) -> ResponseValue: 

243 """Serve file directly from Lektor's output directory.""" 

244 assert isinstance(current_app, LektorApp) 

245 output_path = current_app.lektor_info.output_path 

246 

247 safe_path = safe_join("", *(path.strip("/").split("/"))) 

248 if safe_path is None: 

249 abort(404) 

250 

251 filename = Path(output_path, safe_path) # converts safe_path to native path seps 

252 if filename.is_dir(): 

253 if not path.endswith("/"): 

254 return append_slash_redirect(request.environ) 

255 for index in filename / "index.html", filename / "index.htm": 

256 if index.is_file(): 

257 return _checked_send_file(index, mimetype="text/html") 

258 abort(404) 

259 

260 return _checked_send_file(filename, mimetype=_deduce_mimetype(filename.name)) 

261 

262 

263@bp.route("/", defaults={"path": ""}) 

264@bp.route("/<path:path>") 

265def serve_artifact_or_file(path: str) -> ResponseReturnValue: 

266 try: 

267 return serve_artifact(path) 

268 except HiddenRecordException: 

269 raise 

270 except NotFound: 

271 return serve_file(path) 

272 

273 

274@bp.errorhandler(404) 

275def serve_error_page(error: NotFound) -> ResponseReturnValue: 

276 try: 

277 return serve_artifact("404.html"), 404 

278 except NotFound: 

279 return error