Stream a DM voice message.
(vid: int)
| 13596 | # Let urllib build the redirected request, then validate the target. |
| 13597 | new_req = super().redirect_request(req, fp, code, msg, headers, newurl) |
| 13598 | if new_req is None: |
| 13599 | return None |
| 13600 | _validate_remote_media_url(new_req.full_url) |
| 13601 | return new_req |
| 13602 | |
| 13603 | _safe_urlopen_opener = urllib.request.build_opener(_SafeHTTPRedirectHandler()) |
| 13604 | |
| 13605 | def _safe_urlopen(req: urllib.request.Request, timeout: int = 12): |
| 13606 | """urlopen with strict redirect + host checks (SSRF hardening).""" |
| 13607 | _validate_remote_media_url(req.full_url) |
| 13608 | return _safe_urlopen_opener.open(req, timeout=timeout) |
| 13609 | |
| 13610 | def _dm_store_remote_gif_tx(conn, dm_id: int, gif_url: str) -> Dict[str, Any]: |
| 13611 | """Download and store a remote GIF using an existing DB connection/transaction.""" |
| 13612 | gif_url = (gif_url or "").strip() |
| 13613 | if not gif_url: |
| 13614 | raise ValueError("Empty gif url") |
| 13615 | u = urllib.parse.urlparse(gif_url) |
| 13616 | if u.scheme not in ("http", "https"): |
| 13617 | raise ValueError("GIF url must be http(s)") |
| 13618 | |
| 13619 | _validate_remote_media_url(gif_url) |
| 13620 | |
| 13621 | |
| 13622 | cur = conn.cursor() |
| 13623 | cur.execute( |
| 13624 | "INSERT INTO dm_files(dm_id, filename, mime, stored_path, size, created_at) VALUES(?,?,?,?,?,?)", |
| 13625 | (dm_id, "gif.gif", "image/gif", "PENDING", 0, now_z()) |
| 13626 | ) |
| 13627 | fid = cur.lastrowid |
| 13628 | stored_path = os.path.join(DM_FILES_DIR, f"dm_file_{fid}.bin") |
| 13629 | |
| 13630 | max_bytes = CHAT_MAX_BYTES # 33 MB |
| 13631 | size = 0 |
| 13632 | mime = "image/gif" |
| 13633 | filename = "gif.gif" |
nothing calls this directly
no test coverage detected