Add symbolic link options

py7zr/py7zr.py

@@ -538,6 +538,7 @@ def _extract(
         targets: Optional[Collection[str]] = None,
         return_dict: bool = False,
         callback: Optional[ExtractCallback] = None,
+        enable_symlink: bool = False,
         recursive: Optional[bool] = False,
     ) -> Optional[dict[str, IO[Any]]]:
         if callback is None:
@@ -609,7 +610,12 @@ def _extract(
             elif f.is_socket:
                 pass  # TODO: implement me.
             elif f.is_symlink or f.is_junction:
-                self.worker.register_filelike(f.id, outfilename)
+                if enable_symlink:
+                    self.worker.register_filelike(f.id, outfilename)
+                else:
+                    # Archive has symlink or junction
+                    # this has security consequences.
+                    raise ValueError("Archive has symbolic link that is not explicitly enabled.")
             else:
                 self.worker.register_filelike(f.id, outfilename)
                 target_files.append((outfilename, f.file_properties()))
@@ -713,9 +719,9 @@ def _write_header(self):
         self.sig_header.calccrc(header_len, header_crc)
         self.sig_header.write(self.fp)
 
-    def _writeall(self, path, arcname):
+    def _writeall(self, path, arcname, dereference: bool = False):
         try:
-            if path.is_symlink() and not self.dereference:
+            if path.is_symlink() and not dereference:
                 self.write(path, arcname)
             elif path.is_file():
                 self.write(path, arcname)
@@ -724,14 +730,14 @@ def _writeall(self, path, arcname):
                     self.write(path, arcname)
                 for nm in sorted(os.listdir(str(path))):
                     arc = os.path.join(arcname, nm) if arcname is not None else None
-                    self._writeall(path.joinpath(nm), arc)
+                    self._writeall(path.joinpath(nm), arc, dereference=dereference)
             else:
                 return  # pathlib ignores ELOOP and return False for is_*().
         except OSError as ose:
-            if self.dereference and ose.errno in [errno.ELOOP]:
+            if dereference and ose.errno in [errno.ELOOP]:
                 return  # ignore ELOOP here, this resulted to stop looped symlink reference.
-            elif self.dereference and sys.platform == "win32" and ose.errno in [errno.ENOENT]:
-                return  # ignore ENOENT which is happened when a case of ELOOP on windows.
+            elif dereference and sys.platform == "win32" and ose.errno in [errno.ENOENT]:
+                return  # ignore ENOENT which is happened when a case of ELOOP on Windows.
             else:
                 raise
 
@@ -1020,13 +1026,15 @@ def readall(self) -> Optional[dict[str, IO[Any]]]:
         self._dict = {}
         return self._extract(path=None, return_dict=True)
 
-    def extractall(self, path: Optional[Any] = None, callback: Optional[ExtractCallback] = None) -> None:
+    def extractall(
+        self, path: Optional[Any] = None, callback: Optional[ExtractCallback] = None, enable_symlink: bool = True
+    ) -> None:
         """Extract all members from the archive to the current working
         directory and set owner, modification time and permissions on
         directories afterwards. ``path`` specifies a different directory
         to extract to.
         """
-        self._extract(path=path, return_dict=False, callback=callback)
+        self._extract(path=path, return_dict=False, callback=callback, enable_symlink=enable_symlink)
 
     def read(self, targets: Optional[Collection[str]] = None) -> Optional[dict[str, IO[Any]]]:
         if not self._is_none_or_collection(targets):
@@ -1039,15 +1047,15 @@ def read(self, targets: Optional[Collection[str]] = None) -> Optional[dict[str,
         return self._extract(path=None, targets=targets, return_dict=True)
 
     def extract(
-        self, path: Optional[Any] = None, targets: Optional[Collection[str]] = None, recursive: Optional[bool] = False
-    ) -> None:
+        self, path: Optional[Any] = None, targets: Optional[Collection[str]] = None, recursive: Optional[bool] = False,
+      enable_symlink: bool = True) -> None:
         if not self._is_none_or_collection(targets):
             raise TypeError("Wrong argument type given.")
         # For interoperability with ZipFile, we strip any trailing slashes
         # This also matches the behavior of TarFile
         if targets is not None:
             targets = [remove_trailing_slash(target) for target in targets]
-        self._extract(path, targets, return_dict=False, recursive=recursive)
+        self._extract(path, targets, return_dict=False, recursive=recursive, enable_symlink=enable_symlink)
 
     def reporter(self, callback: ExtractCallback):
         while True:
@@ -1074,18 +1082,18 @@ def reporter(self, callback: ExtractCallback):
                     pass
                 self.q.task_done()
 
-    def writeall(self, path: Union[pathlib.Path, str], arcname: Optional[str] = None):
+    def writeall(self, path: Union[pathlib.Path, str], arcname: Optional[str] = None, dereference: bool = False):
         """Write files in target path into archive."""
         if isinstance(path, str):
             path = pathlib.Path(path)
         if not path.exists():
             raise ValueError("specified path does not exist.")
         if path.is_dir() or path.is_file():
-            self._writeall(path, arcname)
+            self._writeall(path, arcname, dereference or self.dereference)
         else:
             raise ValueError("specified path is not a directory or a file")
 
-    def write(self, file: Union[pathlib.Path, str], arcname: Optional[str] = None):
+    def write(self, file: Union[pathlib.Path, str], arcname: Optional[str] = None, dereference: bool = False):
         """Write single target file into archive."""
         if not isinstance(file, str) and not isinstance(file, pathlib.Path):
             raise ValueError("Unsupported file type.")
@@ -1098,11 +1106,11 @@ def write(self, file: Union[pathlib.Path, str], arcname: Optional[str] = None):
         else:
             path = file
         folder = self.header.initialize()
-        file_info = self._make_file_info(path, arcname, self.dereference)
+        file_info = self._make_file_info(path, arcname, dereference or self.dereference)
         self.header.files_info.files.append(file_info)
         self.header.files_info.emptyfiles.append(file_info["emptystream"])
         self.files.append(file_info)
-        self.worker.archive(self.fp, self.files, folder, deref=self.dereference)
+        self.worker.archive(self.fp, self.files, folder, deref=dereference or self.dereference)
 
     def writed(self, targets: dict[str, IO[Any]]) -> None:
         for target, input in targets.items():