Commit 26bdc6484 for clamav.net
commit 26bdc6484c3d44002d29257fa7c2dba2192ce722
Author: Valerie Snyder <valsnyde@cisco.com>
Date: Wed Aug 5 16:29:04 2026 -0400
Preserve resolved quarantine source paths
FreeBSD F_KINFO resolves a vnode through the name cache and may return a
different hard-link name from the path used to open the scanned file. Using
that result for a path-based quarantine action could remove another link and
leave the submitted path in place.
Retain the already-resolved path supplied to path-based action sources and use
descriptor path resolution only when no authoritative path is available. Guard
F_KINFO use so FreeBSD releases without that command fall back cleanly instead
of failing to compile. Add hard-link regression coverage for move and remove
actions.
CLAM-2959
diff --git a/common/actions.c b/common/actions.c
index fd5921314..4001cf05c 100644
--- a/common/actions.c
+++ b/common/actions.c
@@ -2421,7 +2421,11 @@ static cl_error_t action_source_fallback_action_path_dup(const char *path, char
return action_source_absolute_path_dup(path, action_path);
}
-static cl_error_t action_source_populate_posix(action_source_t *source, int fd, const char *open_path)
+static cl_error_t action_source_populate_posix(
+ action_source_t *source,
+ int fd,
+ const char *open_path,
+ bool open_path_is_resolved)
{
cl_error_t status = CL_EOPEN;
@@ -2436,6 +2440,20 @@ static cl_error_t action_source_populate_posix(action_source_t *source, int fd,
goto done;
}
+ if (open_path_is_resolved) {
+ /*
+ * Descriptor path lookup can return another name for the same file
+ * when it has multiple hard links. Preserve the path that was
+ * explicitly resolved and securely opened for path-based actions.
+ */
+ status = action_source_absolute_path_dup(open_path, &source->action_path);
+ if (CL_SUCCESS == status) {
+ cli_dbgmsg("action_source_populate_posix: Resolved action path for fd [%d] is: %s\n",
+ fd, source->action_path);
+ }
+ goto done;
+ }
+
status = cli_get_filepath_from_filedesc(fd, &source->action_path);
if (CL_SUCCESS != status) {
/*
@@ -2569,7 +2587,7 @@ static cl_error_t action_source_open_path_impl(const char *display_path, const c
goto done;
}
- status = action_source_populate_posix(source, fd, open_path);
+ status = action_source_populate_posix(source, fd, open_path, require_resolved_path);
if (CL_SUCCESS != status) {
goto done;
}
@@ -2650,7 +2668,7 @@ cl_error_t action_source_from_fd(const char *display_path, int fd, action_source
goto done;
}
- status = action_source_populate_posix(source, dup_fd, display_path);
+ status = action_source_populate_posix(source, dup_fd, display_path, false);
if (CL_SUCCESS != status) {
goto done;
}
diff --git a/libclamav/others.h b/libclamav/others.h
index cefcc1a19..92addfb75 100644
--- a/libclamav/others.h
+++ b/libclamav/others.h
@@ -1231,7 +1231,7 @@ cl_error_t cli_get_filepath_from_handle(HANDLE hFile, char **filepath);
* @brief Attempt to get a filename from an open file descriptor.
*
* Caller is responsible for free'ing the filename.
- * Should work on Linux, FreeBSD, macOS, Windows.
+ * Should work on Linux, FreeBSD when F_KINFO is available, macOS, Windows.
*
* @param desc File descriptor
* @param[out] filepath Will be set to file path if found, or NULL.
diff --git a/libclamav/others_common.c b/libclamav/others_common.c
index 8f4cbeef8..0b514953f 100644
--- a/libclamav/others_common.c
+++ b/libclamav/others_common.c
@@ -1496,7 +1496,7 @@ cl_error_t cli_get_filepath_from_filedesc(int desc, char **filepath)
goto done;
}
-#elif defined(__FreeBSD__)
+#elif defined(__FreeBSD__) && defined(F_KINFO)
struct kinfo_file file_info;
diff --git a/unit_tests/clamscan/quarantine_toctou_test.py b/unit_tests/clamscan/quarantine_toctou_test.py
index 59078268a..0e62b6e55 100644
--- a/unit_tests/clamscan/quarantine_toctou_test.py
+++ b/unit_tests/clamscan/quarantine_toctou_test.py
@@ -376,6 +376,62 @@ class TC(testcase.TestCase):
'stdout': completed.stdout,
}
+ def _exercise_source_hardlink_quarantine(self, action_mode: str):
+ assert action_mode in ('move', 'remove')
+
+ parent_dir = TC.path_tmp / ('src-hardlink-{}'.format(action_mode))
+ parent_dir.mkdir()
+
+ db_dir = TC.path_tmp / ('db-src-hardlink-{}'.format(action_mode))
+ db_dir.mkdir()
+
+ payload = b'CLAM-2959 quarantine source hard-link payload\n'
+ other_link_path = parent_dir / 'other-link.bin'
+ other_link_path.write_bytes(payload)
+
+ submitted_path = parent_dir / 'submitted-link.bin'
+ try:
+ os.link(other_link_path, submitted_path)
+ except OSError as err:
+ self.skipTest('Hard-link creation is not permitted in this test environment: {}'.format(err))
+
+ self._write_hdb_signature(db_dir / 'trigger.hdb', payload, 'CLAM-2959-SOURCE-HARDLINK')
+
+ quarantine_dir = None
+ command = []
+ if str(TC.valgrind):
+ command.append(str(TC.valgrind))
+ if TC.valgrind_args:
+ command.extend(TC.valgrind_args.split())
+ command.extend([str(TC.clamscan), '--debug', '-d', str(db_dir)])
+ if action_mode == 'remove':
+ command.append('--remove=yes')
+ else:
+ quarantine_dir = TC.path_tmp / 'quarantine-src-hardlink-move'
+ quarantine_dir.mkdir()
+ command.append('--move={}'.format(quarantine_dir))
+ command.append(str(submitted_path))
+
+ self.log.info('Starting clamscan command: %s', ' '.join(command))
+ completed = subprocess.run(
+ command,
+ cwd=str(TC.path_tmp),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ universal_newlines=True,
+ check=False,
+ )
+ self.log.info('clamscan stdout:\n%s', completed.stdout)
+
+ return {
+ 'payload': payload,
+ 'submitted_path': submitted_path,
+ 'other_link_path': other_link_path,
+ 'quarantine_dir': quarantine_dir,
+ 'returncode': completed.returncode,
+ 'stdout': completed.stdout,
+ }
+
def _exercise_source_link_replacement_quarantine(self, action_mode: str, attempt: int):
assert action_mode in ('copy', 'move', 'remove')
@@ -426,6 +482,7 @@ class TC(testcase.TestCase):
command.append(str(link_path))
milestone_lines = [
+ 'action_source_populate_posix: Resolved action path for fd',
'cli_get_filepath_from_filedesc: File path for fd',
'cli_get_filepath_from_handle: File path for handle',
]
@@ -691,6 +748,32 @@ class TC(testcase.TestCase):
self.assertFalse(payload_path.exists(), 'Expected quarantine removal to unlink the resolved target path.')
self.assertTrue(link_path.is_symlink(), 'Expected the original symlink entry to remain in place after the quarantine remove.')
+ @unittest.skipIf(operating_system == 'windows', 'This test covers POSIX descriptor path resolution.')
+ def test_quarantine_move_unlinks_submitted_hardlink(self):
+ self.step_name('Test quarantine move unlinks the submitted hard-link name')
+ result = self._exercise_source_hardlink_quarantine('move')
+
+ submitted_path = result['submitted_path']
+ other_link_path = result['other_link_path']
+ quarantined_path = result['quarantine_dir'] / submitted_path.name
+
+ self.assertEqual(1, result['returncode'], 'Expected a virus-found exit code from clamscan.')
+ self.assertFalse(submitted_path.exists(), 'Expected quarantine move to unlink the submitted hard-link name.')
+ self.assertTrue(other_link_path.exists(), 'Expected quarantine move to preserve the other hard-link name.')
+ self.assertEqual(result['payload'], other_link_path.read_bytes(), 'Expected the remaining hard link to retain the source bytes.')
+ self.assertTrue(quarantined_path.exists(), 'Expected quarantine move to create the destination file.')
+ self.assertEqual(result['payload'], quarantined_path.read_bytes(), 'Expected the quarantined file to contain the source bytes.')
+
+ @unittest.skipIf(operating_system == 'windows', 'This test covers POSIX descriptor path resolution.')
+ def test_quarantine_remove_unlinks_submitted_hardlink(self):
+ self.step_name('Test quarantine remove unlinks the submitted hard-link name')
+ result = self._exercise_source_hardlink_quarantine('remove')
+
+ self.assertEqual(1, result['returncode'], 'Expected a virus-found exit code from clamscan.')
+ self.assertFalse(result['submitted_path'].exists(), 'Expected quarantine remove to unlink the submitted hard-link name.')
+ self.assertTrue(result['other_link_path'].exists(), 'Expected quarantine remove to preserve the other hard-link name.')
+ self.assertEqual(result['payload'], result['other_link_path'].read_bytes(), 'Expected the remaining hard link to retain the source bytes.')
+
@unittest.skipIf(not hasattr(os, 'symlink'), 'This platform does not support symlink creation in the test environment.')
def test_quarantine_copy_does_not_act_on_replaced_source_link(self):
self.step_name('Test quarantine copy stays bound to the opened source object')