peter
4e4fff87d4
The URL guard judged the address the user typed; a public host's 3xx into our network, a page's formats pointing inside, and a name rebound between the check and the fetch all got past it, and the stored error told a live internal port from a dead one. Every non-YouTube download now runs through an egress gate on the worker's loopback: it resolves each host once, refuses any forbidden answer before dialling, and connects to the address it vetted. Measured on the worker image: yt-dlp's proxy reaches the Requests and CurlCFFI handlers, native HLS and ffmpeg alike. Formats fetched over anything but http(s), which ffmpeg would dial directly, are dropped, and one factory opens every yt-dlp so none skips it.
247 lines
9.8 KiB
Python
247 lines
9.8 KiB
Python
"""The Facebook download path with the user's cookie.
|
|
|
|
Measured on dev (yt-dlp 2026.07.04, the owner's cookie): reel 806379462330494 fails with the cookie
|
|
alone and extracts 2560p with `--impersonate chrome`; the reels that already download with the cookie
|
|
keep their plain first attempt. These pin the attempt order through the real download loop
|
|
(`pipeline._extract_with_fallback`, with yt-dlp faked at the `YoutubeDL` boundary), and that the
|
|
LOCKED curl_cffi can actually impersonate — the pair only works inside yt-dlp's own version range, and
|
|
outside it every target silently reports "unavailable".
|
|
|
|
Measured on dev and in the owner's Chrome: given a group post's link, yt-dlp finds no formats or
|
|
downloads another video from the page's feed. The post's own video is the single Video node in the
|
|
page scripts that carry the post id (prod posts 66, 145, 140, 143, 221; text posts have none while the
|
|
page holds ~10 other videos), and its watch link downloads.
|
|
"""
|
|
|
|
import pytest
|
|
import yt_dlp
|
|
from yt_dlp.networking.exceptions import TransportError
|
|
from yt_dlp.networking.impersonate import ImpersonateTarget
|
|
|
|
from app.downloads import adapters, pipeline
|
|
from app.downloads import errors as dlerrors
|
|
|
|
_URL = "https://www.facebook.com/reel/806379462330494/?fs=e&fs=e"
|
|
_POST = "1755988225525547"
|
|
_GROUP_LINK = f"https://www.facebook.com/groups/983554479435596/permalink/{_POST}/?"
|
|
_POST_VIDEO = "1408015447857753"
|
|
_FEED_VIDEO = "1056963063816966" # what yt-dlp downloaded from this post's link: a feed video
|
|
_CHROME = ImpersonateTarget.from_str("chrome")
|
|
|
|
|
|
def _video_node(video_id: str) -> str:
|
|
# The measured shape of a post's attachment in the page JSON.
|
|
return '{"media":{"__typename":"Video","id":"%s","__isNode":"Video"},"styles":{}}' % video_id
|
|
|
|
|
|
def _page(*post_videos: str, post: str = _POST) -> str:
|
|
attachments = ",".join(_video_node(v) for v in post_videos)
|
|
return (
|
|
'<html><script type="application/json" data-sjs>'
|
|
f'{{"post_id":"{post}","creation_time":1785084554,"attachments":[{attachments}]}}</script>'
|
|
'<script type="application/json" data-sjs>'
|
|
f'{{"feed":[{_video_node(_FEED_VIDEO)}]}}</script></html>'
|
|
)
|
|
|
|
|
|
_WALL = (
|
|
"ERROR: [facebook] 806379462330494: Cannot parse data; please report this issue on "
|
|
"https://github.com/yt-dlp/yt-dlp/issues?q= , filling out the appropriate issue template."
|
|
)
|
|
_NO_FORMATS = "ERROR: [facebook] 2541212983053999: No video formats found!"
|
|
|
|
|
|
def test_the_locked_curl_cffi_can_impersonate_chrome():
|
|
# YoutubeDL refuses to start when the target is unavailable — the same check a download makes.
|
|
with yt_dlp.YoutubeDL({"impersonate": ImpersonateTarget.from_str("chrome"), "quiet": True}):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def ydl(monkeypatch):
|
|
"""Fake `yt_dlp.YoutubeDL`: records each attempt's options and plays back `outcomes` in order;
|
|
`web["page"]` is what a page read returns (`web["fail"]`, when set, is raised instead), and `web`
|
|
records the pages opened, the byte cap each read asked for, and the links extracted."""
|
|
seen: list[dict] = []
|
|
outcomes: list = []
|
|
web: dict = {"page": "", "fail": None, "opened": [], "reads": [], "extracted": []}
|
|
|
|
class FakeYDL:
|
|
def __init__(self, opts):
|
|
seen.append(opts)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def add_post_processor(self, pp, when="post_process"):
|
|
pass # the egress format filter; nothing here to filter
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def urlopen(self, url):
|
|
web["opened"].append(url)
|
|
if web["fail"] is not None:
|
|
raise web["fail"]
|
|
body = web["page"].encode()
|
|
|
|
class Response:
|
|
def read(self, amt=None):
|
|
web["reads"].append(amt)
|
|
return body
|
|
|
|
return Response()
|
|
|
|
def extract_info(self, url, download):
|
|
web["extracted"].append(url)
|
|
outcome = outcomes.pop(0)
|
|
if isinstance(outcome, str):
|
|
raise yt_dlp.utils.DownloadError(outcome)
|
|
return outcome
|
|
|
|
monkeypatch.setattr(pipeline.yt_dlp, "YoutubeDL", FakeYDL)
|
|
monkeypatch.setattr(pipeline, "_touch_job", lambda job_id: None)
|
|
monkeypatch.setattr(pipeline, "_set_job", lambda job_id, **fields: None)
|
|
return seen, outcomes, web
|
|
|
|
|
|
def _run(tmp_path, cookiefile, url=_URL):
|
|
return pipeline._extract_with_fallback(
|
|
1, 1, url, {}, tmp_path, adapters.FacebookAdapter(), cookiefile=cookiefile
|
|
)
|
|
|
|
|
|
def test_a_cookie_download_at_the_wall_is_retried_as_chrome(ydl, tmp_path):
|
|
seen, outcomes, web = ydl
|
|
outcomes += [_WALL, {"id": "806379462330494"}]
|
|
|
|
assert _run(tmp_path, "/cookies.txt") == {"id": "806379462330494"}
|
|
|
|
assert [o.get("impersonate") for o in seen] == [None, _CHROME]
|
|
assert all(o["cookiefile"] == "/cookies.txt" for o in seen) # the retry stays signed in
|
|
assert web["opened"] == [] # a reel is not a group post: no page read
|
|
|
|
|
|
def test_without_a_cookie_the_wall_is_the_login_wall_and_final(ydl, tmp_path):
|
|
seen, outcomes, _ = ydl
|
|
outcomes += [_WALL]
|
|
|
|
with pytest.raises(yt_dlp.utils.DownloadError):
|
|
_run(tmp_path, None)
|
|
|
|
assert len(seen) == 1 and "impersonate" not in seen[0]
|
|
|
|
|
|
def test_other_facebook_failures_are_not_retried_as_chrome(ydl, tmp_path):
|
|
seen, outcomes, _ = ydl
|
|
outcomes += [_NO_FORMATS]
|
|
|
|
with pytest.raises(yt_dlp.utils.DownloadError):
|
|
_run(tmp_path, "/cookies.txt")
|
|
|
|
assert len(seen) == 1 and "impersonate" not in seen[0]
|
|
|
|
|
|
# --- a group post downloads ITS video, or fails saying it has none --------------------------------
|
|
|
|
|
|
def test_a_cookie_group_post_downloads_the_posts_own_video(ydl, tmp_path):
|
|
seen, outcomes, web = ydl
|
|
web["page"] = _page(_POST_VIDEO)
|
|
outcomes += [_WALL, {"id": _POST_VIDEO}] # the watch link needs Chrome too (measured)
|
|
|
|
assert _run(tmp_path, "/cookies.txt", _GROUP_LINK) == {"id": _POST_VIDEO}
|
|
|
|
assert web["opened"] == [_GROUP_LINK]
|
|
# Never an unbounded read.
|
|
assert web["reads"] == [adapters.FacebookAdapter._POST_PAGE_MAX_BYTES]
|
|
assert seen[0]["cookiefile"] == "/cookies.txt" # the post page is read signed in
|
|
watch = f"https://www.facebook.com/watch/?v={_POST_VIDEO}"
|
|
# Never the post link: that is where the feed video comes from.
|
|
assert web["extracted"] == [watch, watch]
|
|
assert [o.get("impersonate") for o in seen[1:]] == [None, _CHROME]
|
|
|
|
|
|
def test_a_group_post_without_a_single_video_fails_saying_so(ydl, tmp_path):
|
|
_, _, web = ydl
|
|
web["page"] = _page() # a text post: the only video on the page is the feed's
|
|
|
|
with pytest.raises(RuntimeError) as exc:
|
|
_run(tmp_path, "/cookies.txt", _GROUP_LINK)
|
|
|
|
assert web["extracted"] == [] # yt-dlp is not left to guess from the post link
|
|
assert dlerrors.diagnose(str(exc.value), url=_GROUP_LINK, cookie_attached=True) == (
|
|
"format_unavailable",
|
|
"facebook.group_post_video",
|
|
)
|
|
|
|
|
|
def test_a_group_post_page_without_the_post_is_the_refused_cookie(ydl, tmp_path):
|
|
# Measured with a refused cookie: Facebook answers the post link with its login page —
|
|
# a login form, no post id anywhere. Saying "this post has no video" would blame the post.
|
|
_, _, web = ydl
|
|
web["page"] = '<html><title>ChatGPT magyarul</title><form id="login_form"></form></html>'
|
|
|
|
with pytest.raises(RuntimeError) as exc:
|
|
_run(tmp_path, "/cookies.txt", _GROUP_LINK)
|
|
|
|
assert web["extracted"] == []
|
|
assert dlerrors.diagnose(str(exc.value), url=_GROUP_LINK, cookie_attached=True) == (
|
|
"login_required",
|
|
"facebook.cookie_rejected",
|
|
)
|
|
|
|
|
|
def test_a_failed_post_page_read_fails_the_download_as_itself(ydl, tmp_path):
|
|
# A read that failed says nothing about the post: it must not turn into "no video in this post".
|
|
_, _, web = ydl
|
|
web["fail"] = TransportError("Connection reset by peer")
|
|
|
|
with pytest.raises(TransportError) as exc:
|
|
_run(tmp_path, "/cookies.txt", _GROUP_LINK)
|
|
|
|
assert web["extracted"] == []
|
|
assert dlerrors.diagnose(str(exc.value), url=_GROUP_LINK, cookie_attached=True) == (
|
|
"network",
|
|
None,
|
|
)
|
|
|
|
|
|
def test_without_a_cookie_the_group_post_link_goes_to_ytdlp_unchanged(ydl, tmp_path):
|
|
_, outcomes, web = ydl
|
|
outcomes += [_NO_FORMATS]
|
|
|
|
with pytest.raises(yt_dlp.utils.DownloadError):
|
|
_run(tmp_path, None, _GROUP_LINK)
|
|
|
|
assert web["opened"] == [] # signed out the page is the login wall: nothing to read
|
|
assert web["extracted"] == [_GROUP_LINK]
|
|
|
|
|
|
def test_the_group_post_link_forms_are_recognised():
|
|
for url in (
|
|
_GROUP_LINK,
|
|
f"https://www.facebook.com/groups/983554479435596/posts/{_POST}/",
|
|
f"https://m.facebook.com/groups/983554479435596/permalink/{_POST}/?",
|
|
f"https://www.facebook.com/groups/chatgptmagyarul/posts/{_POST}",
|
|
):
|
|
assert adapters.facebook_group_post_id(url) == _POST, url
|
|
for url in (
|
|
_URL,
|
|
f"https://www.facebook.com/watch/?v={_POST_VIDEO}",
|
|
f"https://facebook.com.evil/groups/x/permalink/{_POST}/",
|
|
"https://www.facebook.com/groups/983554479435596/",
|
|
):
|
|
assert adapters.facebook_group_post_id(url) is None, url
|
|
assert adapters.SourceAdapter().resolve_url(_GROUP_LINK, {"cookiefile": "/c"}) == _GROUP_LINK
|
|
|
|
|
|
def test_only_the_single_video_of_the_posts_own_blocks_counts():
|
|
assert adapters.facebook_post_video_id(_page(_POST_VIDEO), _POST) == _POST_VIDEO
|
|
# No video in the post, although the page has one elsewhere.
|
|
assert adapters.facebook_post_video_id(_page(), _POST) is None
|
|
# Two videos: guessing between them would be the wrong-video bug again.
|
|
assert adapters.facebook_post_video_id(_page(_POST_VIDEO, "2063302677911049"), _POST) is None
|
|
# A longer number that merely STARTS with the post id is another object.
|
|
assert adapters.facebook_post_video_id(_page(_POST_VIDEO, post=_POST + "0"), _POST) is None
|