28925191300

Committed 08 Jul 2026 07:24AM UTC coverage: 69.222%. First build

Build # 28925191300

Build Type

Pull #1427

github

Committed by

web-flow

Commit Message

Secure user provided values (#1489)

Secure some user provided values in video/utils

Pull Request Pull Request #1427: [Release] 4.3.0

Coverage Stats

915 of 1270 new or added lines in 31 files covered. (72.05%)

13744 of 19855 relevant lines covered (69.22%)

0.69 hits per line

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

75.97

/pod/video_encode_transcript/encoding_utils.py

import json
import logging
import os
import shlex
import subprocess
from collections import OrderedDict
from timeit import default_timer as timer

try:
    from .encoding_settings import VIDEO_RENDITIONS
except (ImportError, ValueError):
    from encoding_settings import VIDEO_RENDITIONS

try:
    from django.conf import settings

    VIDEO_RENDITIONS = getattr(settings, "VIDEO_RENDITIONS", VIDEO_RENDITIONS)
    DEBUG = getattr(settings, "DEBUG", True)
    MEDIA_ROOT = getattr(settings, "MEDIA_ROOT", "")
    DEFAULT_RECORDER_PATH = getattr(settings, "DEFAULT_RECORDER_PATH", "")
except ImportError:  # pragma: no cover
    DEBUG = True
    MEDIA_ROOT = ""
    DEFAULT_RECORDER_PATH = ""

logger = logging.getLogger(__name__)
if DEBUG:
    logger.setLevel(logging.DEBUG)


def sec_to_timestamp(total_seconds) -> str:
    """Format time for webvtt caption."""
    if total_seconds < 0:
        total_seconds = 0
    hours = int(total_seconds // 3600)
    minutes = int((total_seconds % 3600) // 60)
    seconds = total_seconds % 60
    return "{:02d}:{:02d}:{:06.3f}".format(hours, minutes, seconds)


def get_dressing_position_value(position: str, height: str) -> str:
    """
    Obtain dimensions proportional to the video format.

    Args:
        position (str): property "position" of the dressing object.
        height (str): height of the source video.

    Returns:
        str: params for the ffmpeg command.
    """
    height = str(float(height) * 0.05)
    if position == "top_right":
        return "overlay=main_w-overlay_w-" + height + ":" + height
    elif position == "top_left":
        return "overlay=" + height + ":" + height
    elif position == "bottom_right":
        return "overlay=main_w-overlay_w-" + height + ":main_h-overlay_h-" + height
    elif position == "bottom_left":
        return "overlay=" + height + ":main_h-overlay_h-" + height
    return ""


def get_renditions():
    try:
        from django.core import serializers

        from .models import VideoRendition

        renditions = json.loads(
            serializers.serialize("json", VideoRendition.objects.all())
        )
        video_rendition = []
        for rend in renditions:
            video_rendition.append(rend["fields"])
        return video_rendition
    except ImportError:
        return VIDEO_RENDITIONS


def _is_safe_file_path(path_file) -> bool:
    """Return whether a path resolves inside an authorized local root.

    Authorized roots are `MEDIA_ROOT` & `DEFAULT_RECORDER_PATH`.
    """
    if not path_file:
        return False
    try:
        candidate = os.path.abspath(os.path.realpath(os.fspath(path_file)))
    except (TypeError, ValueError, OSError):
        return False

    if not os.path.isabs(candidate):
        return False

    return _is_safe_candidate(candidate, [MEDIA_ROOT, DEFAULT_RECORDER_PATH])


def _is_safe_candidate(candidate, allowed_roots) -> bool:
    """Return whether candidate path is inside any allowed root."""
    for root in allowed_roots:
        if not root:
            continue

        try:
            root_real = os.path.abspath(os.path.realpath(os.fspath(root)))
        except (TypeError, ValueError, OSError):
            continue
        try:
            if os.path.commonpath([candidate, root_real]) == root_real:
                return True
        except ValueError:
            continue
    return False


def check_file(path_file) -> bool:
    """Return whether a path is authorized, existing, and non-empty."""
    if not _is_safe_file_path(path_file):
        return False
    safe_path = os.path.realpath(os.fspath(path_file))
    try:
        return (
            os.path.isfile(safe_path)
            and os.access(safe_path, os.F_OK)
            and os.stat(safe_path).st_size > 0
        )
    except (OSError, ValueError):
        return False


def get_list_rendition():
    list_rendition = {}
    renditions = get_renditions()
    for rend in renditions:
        list_rendition[int(rend["resolution"].split("x")[1])] = rend
    list_rendition = OrderedDict(sorted(list_rendition.items(), key=lambda t: t[0]))
    return list_rendition


def get_info_from_video(probe_cmd):
    info = None
    msg = ""
    try:
        output = subprocess.check_output(shlex.split(probe_cmd), stderr=subprocess.PIPE)
        info = json.loads(output)
    except subprocess.CalledProcessError as e:
        # raise RuntimeError('ffprobe returned non-zero status: {}'.format(
        # e.stderr))
        msg += 20 * "////" + "\n"
        msg += "Runtime Error: {0}\n".format(e)
    except OSError as err:
        # raise OSError(e.errno, 'ffprobe not found: {}'.format(e.strerror))
        msg += 20 * "////" + "\n"
        msg += "OS error: {0}\n".format(err)
    return info, msg


def launch_cmd(cmd):
    if cmd == "":
        msg = "No cmd to launch"
        logger.warning(msg)
        return False, msg
    msg = ""
    encode_start = timer()
    return_value = False
    try:
        logger.debug("launch_cmd: %s" % cmd)
        output = subprocess.run(
            shlex.split(cmd),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
        encode_end = timer() - encode_start
        msg += cmd + "\n"
        msg += "Encode file in {:.3}s.\n".format(encode_end)
        # msg += "\n".join(output.stdout.decode().split('\n'))
        try:
            msg += output.stdout.decode("utf-8")
        except UnicodeDecodeError:
            pass
        msg += "\n"
        if output.returncode != 0:
            msg += "ERROR RETURN CODE %s for command %s" % (output.returncode, cmd)
        else:
            return_value = True
    except (subprocess.CalledProcessError, OSError) as e:
        # raise RuntimeError('ffmpeg returned non-zero status: {}'.format(
        # e.stderr))
        msg += 20 * "////" + "\n"
        err_msg = "Runtime or OS Error: {0}\n".format(e)
        msg += err_msg
        logger.error(err_msg)
    return return_value, msg

1	import json	1✔
2	import logging	1✔
3	import os	1✔
4	import shlex	1✔
5	import subprocess	1✔
6	from collections import OrderedDict	1✔
7	from timeit import default_timer as timer	1✔
8
9	try:	1✔
10	from .encoding_settings import VIDEO_RENDITIONS	1✔
11	except (ImportError, ValueError):	×
12	from encoding_settings import VIDEO_RENDITIONS	×
13
14	try:	1✔
15	from django.conf import settings	1✔
16
17	VIDEO_RENDITIONS = getattr(settings, "VIDEO_RENDITIONS", VIDEO_RENDITIONS)	1✔
18	DEBUG = getattr(settings, "DEBUG", True)	1✔
19	MEDIA_ROOT = getattr(settings, "MEDIA_ROOT", "")	1✔
20	DEFAULT_RECORDER_PATH = getattr(settings, "DEFAULT_RECORDER_PATH", "")	1✔
21	except ImportError: # pragma: no cover
22	DEBUG = True
23	MEDIA_ROOT = ""
24	DEFAULT_RECORDER_PATH = ""
25
26	logger = logging.getLogger(__name__)	1✔
27	if DEBUG:	1✔
28	logger.setLevel(logging.DEBUG)	1✔
29
30
31	def sec_to_timestamp(total_seconds) -> str:	1✔
32	"""Format time for webvtt caption."""
33	if total_seconds < 0:	1✔
34	total_seconds = 0	1✔
35	hours = int(total_seconds // 3600)	1✔
36	minutes = int((total_seconds % 3600) // 60)	1✔
37	seconds = total_seconds % 60	1✔
38	return "{:02d}:{:02d}:{:06.3f}".format(hours, minutes, seconds)	1✔
39
40
41	def get_dressing_position_value(position: str, height: str) -> str:	1✔
42	"""
43	Obtain dimensions proportional to the video format.
44
45	Args:
46	position (str): property "position" of the dressing object.
47	height (str): height of the source video.
48
49	Returns:
50	str: params for the ffmpeg command.
51	"""
52	height = str(float(height) * 0.05)	1✔
53	if position == "top_right":	1✔
54	return "overlay=main_w-overlay_w-" + height + ":" + height	1✔
55	elif position == "top_left":	1✔
56	return "overlay=" + height + ":" + height	1✔
57	elif position == "bottom_right":	1✔
58	return "overlay=main_w-overlay_w-" + height + ":main_h-overlay_h-" + height	1✔
59	elif position == "bottom_left":	1✔
60	return "overlay=" + height + ":main_h-overlay_h-" + height	1✔
61	return ""	×
62
63
64	def get_renditions():	1✔
65	try:	1✔
66	from django.core import serializers	1✔
67
68	from .models import VideoRendition	1✔
69
70	renditions = json.loads(	1✔
71	serializers.serialize("json", VideoRendition.objects.all())
72	)
73	video_rendition = []	1✔
74	for rend in renditions:	1✔
75	video_rendition.append(rend["fields"])	1✔
76	return video_rendition	1✔
77	except ImportError:	×
78	return VIDEO_RENDITIONS	×
79
80
81	def _is_safe_file_path(path_file) -> bool:	1✔
82	"""Return whether a path resolves inside an authorized local root.
83
84	Authorized roots are `MEDIA_ROOT` & `DEFAULT_RECORDER_PATH`.
85	"""
86	if not path_file:	1✔
87	return False	1✔
88	try:	1✔
89	candidate = os.path.abspath(os.path.realpath(os.fspath(path_file)))	1✔
NEW 90	except (TypeError, ValueError, OSError):	×
NEW 91	return False	×
92
93	if not os.path.isabs(candidate):	1✔
NEW 94	return False	×
95
96	return _is_safe_candidate(candidate, [MEDIA_ROOT, DEFAULT_RECORDER_PATH])	1✔
97
98
99	def _is_safe_candidate(candidate, allowed_roots) -> bool:	1✔
100	"""Return whether candidate path is inside any allowed root."""
101	for root in allowed_roots:	1✔
102	if not root:	1✔
103	continue	1✔
104
105	try:	1✔
106	root_real = os.path.abspath(os.path.realpath(os.fspath(root)))	1✔
NEW 107	except (TypeError, ValueError, OSError):	×
NEW 108	continue	×
109	try:	1✔
110	if os.path.commonpath([candidate, root_real]) == root_real:	1✔
111	return True	1✔
NEW 112	except ValueError:	×
NEW 113	continue	×
114	return False	1✔
115
116
117	def check_file(path_file) -> bool:	1✔
118	"""Return whether a path is authorized, existing, and non-empty."""
119	if not _is_safe_file_path(path_file):	1✔
120	return False	1✔
121	safe_path = os.path.realpath(os.fspath(path_file))	1✔
122	try:	1✔
123	return (	1✔
124	os.path.isfile(safe_path)
125	and os.access(safe_path, os.F_OK)
126	and os.stat(safe_path).st_size > 0
127	)
NEW 128	except (OSError, ValueError):	×
NEW 129	return False	×
130
131
132	def get_list_rendition():	1✔
133	list_rendition = {}	1✔
134	renditions = get_renditions()	1✔
135	for rend in renditions:	1✔
136	list_rendition[int(rend["resolution"].split("x")[1])] = rend	1✔
137	list_rendition = OrderedDict(sorted(list_rendition.items(), key=lambda t: t[0]))	1✔
138	return list_rendition	1✔
139
140
141	def get_info_from_video(probe_cmd):	1✔
142	info = None	1✔
143	msg = ""	1✔
144	try:	1✔
145	output = subprocess.check_output(shlex.split(probe_cmd), stderr=subprocess.PIPE)	1✔
146	info = json.loads(output)	1✔
147	except subprocess.CalledProcessError as e:	×
148	# raise RuntimeError('ffprobe returned non-zero status: {}'.format(
149	# e.stderr))
150	msg += 20 * "////" + "\n"	×
151	msg += "Runtime Error: {0}\n".format(e)	×
152	except OSError as err:	×
153	# raise OSError(e.errno, 'ffprobe not found: {}'.format(e.strerror))
154	msg += 20 * "////" + "\n"	×
155	msg += "OS error: {0}\n".format(err)	×
156	return info, msg	1✔
157
158
159	def launch_cmd(cmd):	1✔
160	if cmd == "":	1✔
161	msg = "No cmd to launch"	×
162	logger.warning(msg)	×
163	return False, msg	×
164	msg = ""	1✔
165	encode_start = timer()	1✔
166	return_value = False	1✔
167	try:	1✔
168	logger.debug("launch_cmd: %s" % cmd)	1✔
169	output = subprocess.run(	1✔
170	shlex.split(cmd),
171	stdout=subprocess.PIPE,
172	stderr=subprocess.STDOUT,
173	)
174	encode_end = timer() - encode_start	1✔
175	msg += cmd + "\n"	1✔
176	msg += "Encode file in {:.3}s.\n".format(encode_end)	1✔
177	# msg += "\n".join(output.stdout.decode().split('\n'))
178	try:	1✔
179	msg += output.stdout.decode("utf-8")	1✔
180	except UnicodeDecodeError:	×
181	pass	×
182	msg += "\n"	1✔
183	if output.returncode != 0:	1✔
184	msg += "ERROR RETURN CODE %s for command %s" % (output.returncode, cmd)	×
185	else:
186	return_value = True	1✔
187	except (subprocess.CalledProcessError, OSError) as e:	×
188	# raise RuntimeError('ffmpeg returned non-zero status: {}'.format(
189	# e.stderr))
190	msg += 20 * "////" + "\n"	×
191	err_msg = "Runtime or OS Error: {0}\n".format(e)	×
192	msg += err_msg	×
193	logger.error(err_msg)	×
194	return return_value, msg	1✔

EsupPortail / Esup-Pod / 28925191300

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