Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix symlink to same exterior location #1167

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix symlink to same exterior location #1167

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
41321a1
Fix another cornercase with symlink handling
abadger Jan 19, 2024
bc23ec0
Fix a cornercase in choosing whether to copy or link a file.
abadger Jan 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 129 additions & 60 deletions repos/system_upgrade/common/actors/targetuserspacecreator/libraries/userspacegen.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -348,9 +348,65 @@ def _mkdir_with_copied_mode(path, mode_from):
run(['chmod', '--reference={}'.format(mode_from), path])


def _canonize(pointee_as_abspath, srcdir):
"""
Helper function to return canonical forms of directory paths.

This is an ugly function that just avoids having to duplicate this code in two places.
"""
# If srcdir is a symlink, then we need a name for it that we can compare
# with other paths.
canonical_srcdir = os.path.realpath(srcdir)

# To make comparisons, we need to resolve all symlinks in the directory
# structure leading up to pointee. However, we can't include pointee
# itself otherwise it will resolve to the file that it points to in the
# end (which would be wrong if pointee_filename is a symlink).
canonical_pointee_dir, pointee_filename = os.path.split(pointee_as_abspath)
canonical_pointee_dir = os.path.realpath(canonical_pointee_dir)

return (pointee_filename, canonical_pointee_dir, canonical_srcdir)


def _format_source_link(pointee_as_abspath, symlink, srcdir):
"""
Format the source for a symlink as either an absolute path or a path relative to srcdir.

:param pointee_as_abspath: Sbsolute path to the source of the symlink.
:param symlink: Absolute path to the symlink that started us looking
through the chain of links. If the symlink points to a relative path, this function will
return a relative path. If it is an absolute path, it will return an absolute path.
:param srcdir: The directory in the source tree that we want everything to live within.
"""
pointee_filename, canonical_pointee_dir, canonical_srcdir = _canonize(pointee_as_abspath, srcdir)
link_to = os.readlink(symlink)

if link_to.startswith('/'):
# The original symlink was an absolute path so we will set this
# one to absolute too
# Note: Because absolute paths are constructed inside of
# srcdir, the relative path that we need to join with has to be
# relative to srcdir, not the directory that the symlink is
# being created in.
relative_to_srcdir = os.path.relpath(canonical_pointee_dir, canonical_srcdir)
corrected_path = os.path.normpath(os.path.join(srcdir,
relative_to_srcdir,
pointee_filename)
)
else:
# If the original link is a relative link, then we want the new
# link to be relative (to the path that we will place the symlink in) as well
canonical_symlink_dir = os.path.realpath(os.path.dirname(symlink))
relative_to_symlink_location = os.path.relpath(canonical_pointee_dir, canonical_symlink_dir)
corrected_path = os.path.normpath(os.path.join(relative_to_symlink_location,
pointee_filename)
)
return corrected_path


def _choose_copy_or_link(symlink, srcdir):
"""
Copy file contents or create a symlink depending on where the pointee resides.
Determine whether to copy file contents or create a symlink depending on where the pointee resides.

:param symlink: The source symlink to follow. This must be an absolute path.
:param srcdir: The root directory that every piece of content must be present in.
Expand Down Expand Up @@ -378,10 +434,6 @@ def _choose_copy_or_link(symlink, srcdir):
if not os.path.exists(symlink):
raise BrokenSymlinkError('File {} is a broken symlink!'.format(symlink))

# If srcdir is a symlink, then we need a name for it that we can compare
# with other paths.
canonical_srcdir = os.path.realpath(srcdir)

pointee_as_abspath = symlink
seen = set([pointee_as_abspath])

Expand Down Expand Up @@ -412,48 +464,70 @@ def _choose_copy_or_link(symlink, srcdir):

seen.add(pointee_as_abspath)

# To make comparisons, we need to resolve all symlinks in the directory
# structure leading up to pointee. However, we can't include pointee
# itself otherwise it will resolve to the file that it points to in the
# end.
canonical_pointee_dir, pointee_filename = os.path.split(pointee_as_abspath)
canonical_pointee_dir = os.path.realpath(canonical_pointee_dir)

if canonical_pointee_dir.startswith(canonical_srcdir):
# Absolute path inside of the correct dir so we need to link to it
# But we need to determine what the link path should be before
# returning.

# Construct a relative path that points from the symlinks directory
# to the pointee.
link_to = os.readlink(symlink)
canonical_symlink_dir = os.path.realpath(os.path.dirname(symlink))
relative_path = os.path.relpath(canonical_pointee_dir, canonical_symlink_dir)

if link_to.startswith('/'):
# The original symlink was an absolute path so we will set this
# one to absolute too
# Note: Because absolute paths are constructed inside of
# srcdir, the relative path that we need to join here has to be
# relative to srcdir, not the directory that the symlink is
# being created in.
relative_to_srcdir = os.path.relpath(canonical_pointee_dir, canonical_srcdir)
corrected_path = os.path.normpath(os.path.join(srcdir, relative_to_srcdir, pointee_filename))

else:
# If the original link is a relative link, then we want the new
# link to be relative as well
corrected_path = os.path.normpath(os.path.join(relative_path, pointee_filename))

return ("link", corrected_path)
dummy_pointee_filename, canonical_pointee_dir, canonical_srcdir = _canonize(pointee_as_abspath, srcdir)
if not canonical_pointee_dir.startswith(canonical_srcdir):
# pointee is a symlink that points outside of the srcdir so continue to
# the next symlink in the chain.
continue

# pointee is a symlink that points outside of the srcdir so continue to
# the next symlink in the chain.
# Absolute path inside of the correct dir so we need to link to it
# Note: We have to make the source of the link relative or absolute
# depending on what the original link was
corrected_path = _format_source_link(pointee_as_abspath, symlink, srcdir)
return ("link", corrected_path)

# The file is not a link so copy it
return ('copy', pointee_as_abspath)


def _copy_symlinks(symlinks_to_process, srcdir):
"""
Copy file contents or create a symlink depending on where the pointee resides.

:param symlinks_to_process: List of 2-tuples of (src_path, target_path). Each src_path
should be an absolute path to the symlink. target_path is the path to where we
need to create either a link or a copy.
:param srcdir: The root directory that every piece of content must be present in.
:raises ValueError: if the arguments are not correct
"""
copied_links = {}
for source_linkpath, target_linkpath in symlinks_to_process:
try:
action, source_path = _choose_copy_or_link(source_linkpath, srcdir)
except BrokenSymlinkError as e:
# Skip and report broken symlinks
api.current_logger().warning('{} Will not copy the file!'.format(str(e)))
continue

if action == "copy":
# Need the canonical path so that we can check if the pointed to
# file has been copied before
canonical_source_path = os.path.realpath(source_path)
if canonical_source_path in copied_links:
# The link was copied before so we need to change this to a link action
# with the previous copied link location as the source for this link
action = "link"
source_abspath = os.path.abspath(os.path.normpath(copied_links[canonical_source_path]))
source_path = _format_source_link(source_abspath, source_linkpath, srcdir)
else:
# All other links to the same target will need to link to the path
# that we're processing now since they all pointed to the same
# target before.
copied_links[canonical_source_path] = source_linkpath

# Note: source_path could be a directory, so '-a' or '-r' must be
# given to cp.
run(['cp', '-a', source_path, target_linkpath])
continue

if action == 'link':
run(["ln", "-s", source_path, target_linkpath])
continue

# This will not happen unless _copy_or_link() has a bug.
raise RuntimeError("Programming error: _copy_or_link() returned an unknown action:{}".format(action))


def _copy_decouple(srcdir, dstdir):
"""
Copy files inside of `srcdir` to `dstdir` while decoupling symlinks.
Expand All @@ -467,7 +541,6 @@ def _copy_decouple(srcdir, dstdir):
.. warning::
`dstdir` must already exist.
"""
symlinks_to_process = []
for root, directories, files in os.walk(srcdir):
# relative path from srcdir because srcdir is replaced with dstdir for
# the copy.
Expand All @@ -476,11 +549,24 @@ def _copy_decouple(srcdir, dstdir):
# Create all directories with proper permissions for security
# reasons (Putting private data into directories that haven't had their
# permissions set appropriately may leak the private information.)
symlinks_to_process = []
for directory in directories:
source_dirpath = os.path.join(root, directory)
target_dirpath = os.path.join(dstdir, relpath, directory)

# Defer symlinks until later because we may end up having to copy
# the file contents and the directory may not exist yet.
if os.path.islink(source_dirpath):
symlinks_to_process.append((source_dirpath, target_dirpath))
continue

_mkdir_with_copied_mode(target_dirpath, source_dirpath)

# Link or create all directories that were pointed to by symlinks and
# then reset symlinks_to_process for use by files.
_copy_symlinks(symlinks_to_process, srcdir)
symlinks_to_process = []

for filename in files:
source_filepath = os.path.join(root, filename)
target_filepath = os.path.join(dstdir, relpath, filename)
Expand All @@ -494,24 +580,7 @@ def _copy_decouple(srcdir, dstdir):
# Not a symlink so we can copy it now too
run(['cp', '-a', source_filepath, target_filepath])

# Now process all symlinks
for source_linkpath, target_linkpath in symlinks_to_process:
try:
action, source_path = _choose_copy_or_link(source_linkpath, srcdir)
except BrokenSymlinkError as e:
# Skip and report broken symlinks
api.current_logger().warning('{} Will not copy the file!'.format(str(e)))
continue

if action == "copy":
# Note: source_path could be a directory, so '-a' or '-r' must be
# given to cp.
run(['cp', '-a', source_path, target_linkpath])
elif action == 'link':
run(["ln", "-s", source_path, target_linkpath])
else:
# This will not happen unless _copy_or_link() has a bug.
raise RuntimeError("Programming error: _copy_or_link() returned an unknown action:{}".format(action))
_copy_symlinks(symlinks_to_process, srcdir)


def _copy_certificates(context, target_userspace):
Expand Down
118 changes: 118 additions & 0 deletions ...em_upgrade/common/actors/targetuserspacecreator/tests/unit_test_targetuserspacecreator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -381,6 +381,65 @@ def temp_directory_layout(tmp_path, initial_structure):
},
id="Absolute_symlink_to_a_file_inside_via_a_symlink_to_the_rootdir"
)),
(pytest.param(
{
'dir': {
'fileA': '/outside/fileC',
'fileB': '/outside/fileC',
},
'outside': {
'fileC': None,
},
},
{
'dir': {
'fileA': None,
'fileB': '/dir/fileA',
},
},
id="Absolute_two_symlinks_to_the_same_copied_file"
)),
(pytest.param(
{
'dir': {
'fileA': None,
'link_to_dir': '/dir/inside',
'inside': {
'fileB': None,
},
},
},
{
'dir': {
'fileA': None,
'link_to_dir': '/dir/inside',
'inside': {
'fileB': None,
},
},
},
id="Absolute_symlink_to_a_dir_inside"
)),
(pytest.param(
{
'dir': {
'fileA': None,
'link_to_dir': '/outside',
},
'outside': {
'fileB': None,
},
},
{
'dir': {
'fileA': None,
'link_to_dir': {
'fileB': None,
},
},
},
id="Absolute_symlink_to_a_dir_outside"
)),
(pytest.param(
# This one is very tricky:
# * The user has made /etc/pki a symlink to some other directory that
Expand Down Expand Up @@ -671,6 +730,65 @@ def temp_directory_layout(tmp_path, initial_structure):
},
id="Relative_symlink_to_a_file_inside_via_a_symlink_to_the_rootdir"
)),
(pytest.param(
{
'dir': {
'fileA': '../outside/fileC',
'fileB': '../outside/fileC',
},
'outside': {
'fileC': None,
},
},
{
'dir': {
'fileA': None,
'fileB': 'fileA',
},
},
id="Relative_two_symlinks_to_the_same_copied_file"
)),
(pytest.param(
{
'dir': {
'fileA': None,
'link_to_dir': '../outside',
},
'outside': {
'fileB': None,
},
},
{
'dir': {
'fileA': None,
'link_to_dir': {
'fileB': None,
},
},
},
id="Relative_symlink_to_a_dir_outside"
)),
(pytest.param(
{
'dir': {
'fileA': None,
'link_to_dir': 'inside',
'inside': {
'fileB': None,
},
},
},
{
'dir': {
'fileA': None,
'link_to_dir': 'inside',
'inside': {
'fileB': None,
},
},
},
id="Relative_symlink_to_a_dir_inside"
)),
(pytest.param(
# This one is very tricky:
# * The user has made /etc/pki a symlink to some other directory that
Expand Down
Loading