Fix workload socket denied (#5698)

Fix for #5672, #5505 , #5693

Restarting iotedge many time lead to the following permission issues:
![image](https://user-images.githubusercontent.com/33438817/137411625-b2eafa34-979a-4d10-a022-862f1dc84e25.png)

This doesn't seem to happen in 1.1.6. I tried many times but could not get it to fail however some customer experienced the same symptoms so this is most likely an issue across version.
The fact that it seems to fail more on some setup is something we could not explain.

Tests:
1.  Tried all of those. A few times manually (3-5 times), a cycle of hundreds time each with a script (Ubuntu + Centos).
For each test we check that all module are up and running, that permissions and user are correct and that there is a listener on the socket:
1.1 sudo iotedge system restart
1.2 sudo iotedge system stop + delete all containers + sudo iotedge system restart
1.3 sudo iotedge system stop + delete /var/lib/aziot/edged/mnt/ folder + sudo iotedge system restart  
1.4 sudo iotedge system stop + delete all container + delete /var/lib/aziot/edged/mnt/ folder + sudo iotedge system restart  
1.5 sudo iotedge system stop + delete workloads inside /var/lib/aziot/edged/mnt + create a sudo dir inside with the name of the sockets + sudo iotedge system restart 

Logs
[ubuntu18.txt](https://github.com/Azure/iotedge/files/7358293/ubuntu18.txt)
[centos.txt](https://github.com/Azure/iotedge/files/7358294/centos.txt)

scripts code:
```
while :
do
  echo "Test 1"
  sudo iotedge system restart
  sleep 310s
  sudo iotedge list
  ls -l /var/lib/aziot/edged/mnt/
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeAgent.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeHub.sock http://127.0.0.1


  echo "Test 2"
  sudo iotedge system stop
  sudo docker rm -f edgeAgent
  sudo docker rm -f edgeHub
  sudo docker rm -f SimulatedTemperatureSensor
  sudo iotedge system restart
  sleep 120s
  sudo iotedge list
  ls -l /var/lib/aziot/edged/mnt/
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeAgent.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeHub.sock http://127.0.0.1

  echo "Test 3"
  sudo iotedge system stop
  sudo rm -r /var/lib/aziot/edged/mnt
  sudo iotedge system restart
  sleep 310s
  sudo iotedge list
  ls -l /var/lib/aziot/edged/mnt/
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeAgent.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeHub.sock http://127.0.0.1

  echo "Test 4"
  sudo iotedge system stop
  sudo docker rm -f edgeAgent
  sudo docker rm -f edgeHub
  sudo docker rm -f SimulatedTemperatureSensor
  sudo rm -r /var/lib/aziot/edged/mnt
  sudo iotedge system restart
  sleep 120s
  sudo iotedge list
  ls -l /var/lib/aziot/edged/mnt/
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeAgent.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeHub.sock http://127.0.0.1

  echo "Test 5"
  sudo iotedge system stop
  sudo rm /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock
  sudo rm /var/lib/aziot/edged/mnt/edgeAgent.sock
  sudo rm /var/lib/aziot/edged/mnt/edgeHub.sock
  sudo mkdir /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock
  sudo mkdir /var/lib/aziot/edged/mnt/edgeAgent.sock
  sudo mkdir /var/lib/aziot/edged/mnt/edgeHub.sock
  sudo iotedge system restart
  sleep 310s
  sudo iotedge list
  ls -l /var/lib/aziot/edged/mnt/
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/SimulatedTemperatureSensor.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeAgent.sock http://127.0.0.1
  curl curl --unix-socket  /var/lib/aziot/edged/mnt/edgeHub.sock http://127.0.0.1

done

```



2. Pipeline run:
Package: 47895915
Run https://dev.azure.com/msazure/One/_build/results?buildId=47896816&view=results

5. Test the protection against 2 listeners: Crashed edgeAgent to have edged restart it. EdgeAgent doesn't get stopped before starting like other module: 
Listener  EdgeAgent already started, removing old listener.
Confirmed that 2 listeners is not a problem.



# Azure IoT Edge PR checklist:

This checklist is used to make sure that common guidelines for a pull request are followed.

### General Guidelines and Best Practices
- [x] I have read the [contribution guidelines](https://github.com/azure/iotedge#contributing).
- [x] Title of the pull request is clear and informative.
- [x] Description of the pull request includes a concise summary of the enhancement or bug fix.

### Testing Guidelines
- [x] Pull request includes test coverage for the included changes.
- Description of the pull request includes 
	- [ ] concise summary of tests added/modified
	- [x] local testing done.  

### Draft PRs
- Open the PR in `Draft` mode if it is:
	- Work in progress or not intended to be merged.
	- Encountering multiple pipeline failures and working on fixes.

_Note: We use the kodiakhq bot to merge PRs once the necessary checks and approvals are in place. When it merges a PR, kodiakhq converts the PR title to the commit title, PR description to the commit description, and squashes all the commits in the PR to a single commit. The net effect is that entire PR becomes a single commit. Please follow the best practices mentioned [here](https://chris.beams.io/posts/git-commit/#:~:text=The%20seven%20rules%20of%20a%20great%20Git%20commit,what%20and%20why%20vs.%20how%20For%20example%3A%20) for the PR title and description_

edgelet/aziot-edged/src/workload_manager.rs

@@ -126,6 +126,25 @@ where
     ) -> Result<(), Error> {
         let label = "work".to_string();
 
+        // If a listener has already been created, remove previous listener.
+        // This avoid the launch of 2 listeners.
+        // We chose to remove instead and create a new one instead of
+        // just return and say, one listener has already been created:
+        // We chose to remove because a listener could crash without getting removed correctly.
+        // That could make the module crash. Then that module would be restarted without ever going to
+        // "stop"
+        // There is still a chance that 2 concurrent servers are launch with concurrence,
+        // But it is extremely unlikely and anyway doesn't have any side effect expect memory footprint.
+        if let Some(shutdown_sender) = self.shutdown_senders.remove(module_id) {
+            info!(
+                "Listener  {} already started, removing old listener",
+                module_id
+            );
+            shutdown_sender
+                .send(())
+                .map_err(|()| Error::from(ErrorKind::WorkloadService))?;
+        }
+
         let (shutdown_sender, shutdown_receiver) = oneshot::channel();
 
         self.shutdown_senders

edgelet/edgelet-http/Cargo.toml

@@ -34,7 +34,6 @@ systemd = { path = "../systemd" }
 hyperlocal = "0.6"
 libc = "0.2"
 nix = "0.14"
-scopeguard = "0.3.3"
 tokio-uds = "0.2"
 
 [dev-dependencies]

edgelet/edgelet-http/src/unix.rs

@@ -1,106 +1,61 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 use std::fs;
-#[cfg(unix)]
-use std::os::unix::fs::MetadataExt;
 use std::os::unix::prelude::PermissionsExt;
 use std::path::Path;
 
 use failure::ResultExt;
 use log::{debug, error};
 #[cfg(unix)]
-use nix::sys::stat::{umask, Mode};
-#[cfg(unix)]
-use scopeguard::defer;
-#[cfg(unix)]
 use tokio_uds::UnixListener;
 
 use crate::error::{Error, ErrorKind};
 use crate::util::{incoming::Incoming, socket_file_exists};
 
 pub fn listener<P: AsRef<Path>>(path: P, unix_socket_permission: u32) -> Result<Incoming, Error> {
-    let listener = if socket_file_exists(path.as_ref()) {
-        // get the previous file's metadata
-        #[cfg(unix)]
-        let metadata = get_metadata(path.as_ref())?;
-
-        debug!("unlinking {}...", path.as_ref().display());
-        fs::remove_file(&path)
-            .with_context(|_| ErrorKind::Path(path.as_ref().display().to_string()))?;
-        debug!("unlinked {}", path.as_ref().display());
-
-        #[cfg(unix)]
-        let prev = set_umask(&metadata, path.as_ref());
-        #[cfg(unix)]
-        defer! {{ umask(prev); }}
-
-        debug!("binding {}...", path.as_ref().display());
-
-        let listener = UnixListener::bind(&path)
-            .with_context(|_| ErrorKind::Path(path.as_ref().display().to_string()))?;
-        debug!("bound {}", path.as_ref().display());
-
-        Incoming::Unix(listener)
-    } else {
-        // If parent doesn't exist, create it and socket will be created inside.
-        if let Some(parent) = path.as_ref().parent() {
-            if !parent.exists() {
-                fs::create_dir_all(parent).with_context(|err| {
-                    error!("Cannot create directory, error: {}", err);
-                    ErrorKind::Path(path.as_ref().display().to_string())
-                })?;
-            }
-        }
-
-        let listener = UnixListener::bind(&path)
-            .with_context(|_| ErrorKind::Path(path.as_ref().display().to_string()))?;
-
-        fs::set_permissions(
-            path.as_ref(),
-            fs::Permissions::from_mode(unix_socket_permission),
-        )
-        .map_err(|err| {
-            error!("Cannot set directory permissions: {}", err);
-            ErrorKind::Path(path.as_ref().display().to_string())
-        })?;
+    let path = path.as_ref();
+    let path_display = path.display();
 
-        Incoming::Unix(listener)
-    };
+    if socket_file_exists(path) {
+        debug!("unlinking {}...", path_display);
 
-    Ok(listener)
-}
-
-#[cfg(unix)]
-fn get_metadata(path: &Path) -> Result<fs::Metadata, Error> {
-    let metadata =
-        fs::metadata(path).with_context(|_| ErrorKind::Path(path.display().to_string()))?;
-    debug!("read metadata {:?} for {}", metadata, path.display());
-    Ok(metadata)
-}
-
-#[cfg(unix)]
-fn set_umask(metadata: &fs::Metadata, path: &Path) -> Mode {
-    #[cfg(target_os = "macos")]
-    #[allow(clippy::cast_possible_truncation)]
-    let mode = Mode::from_bits_truncate(metadata.mode() as u16);
+        let err1 = fs::remove_file(path).err();
+        let err2 = fs::remove_dir_all(path).err();
+        if let Some((err1, err2)) = err1.zip(err2) {
+            error!("Could not unlink existing socket: [{}] [{}]", err1, err2);
+            return Err(ErrorKind::Path(path_display.to_string()).into());
+        }
+        debug!("unlinked {}", path_display);
+    }
 
-    #[cfg(not(target_os = "macos"))]
-    let mode = Mode::from_bits_truncate(metadata.mode());
+    // If parent doesn't exist, create it and socket will be created inside.
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent).with_context(|err| {
+            error!("Cannot create directory, error: {}", err);
+            ErrorKind::Path(path_display.to_string())
+        })?;
+    }
 
-    let mut mask = Mode::all();
-    mask.toggle(mode);
+    let listener =
+        UnixListener::bind(&path).with_context(|_| ErrorKind::Path(path_display.to_string()))?;
+    debug!("bound {}", path_display);
 
-    debug!("settings permissions {:#o} for {}...", mode, path.display());
+    fs::set_permissions(path, fs::Permissions::from_mode(unix_socket_permission)).map_err(
+        |err| {
+            error!("Cannot set directory permissions: {}", err);
+            ErrorKind::Path(path_display.to_string())
+        },
+    )?;
 
-    umask(mask)
+    Ok(Incoming::Unix(listener))
 }
 
 #[cfg(test)]
 #[cfg(unix)]
 mod tests {
-    use super::{listener, MetadataExt};
-
+    use super::listener;
     use std::fs::OpenOptions;
+    use std::os::unix::fs::MetadataExt;
     use std::os::unix::fs::OpenOptionsExt;
 
     use futures::Stream;
@@ -126,7 +81,21 @@ mod tests {
         let _srv = listener.for_each(move |(_socket, _addr)| Ok(()));
 
         let file_stat = stat(&path).unwrap();
-        assert_eq!(0o600, file_stat.st_mode & 0o777);
+        assert_eq!(0o666, file_stat.st_mode & 0o777);
+
+        dir.close().unwrap();
+    }
+
+    #[test]
+    fn test_socket_not_created() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("dummy_socket.sock");
+
+        let listener = listener(&path, 0o666).unwrap();
+        let _srv = listener.for_each(move |(_socket, _addr)| Ok(()));
+
+        let file_stat = stat(&path).unwrap();
+        assert_eq!(0o666, file_stat.st_mode & 0o777);
 
         dir.close().unwrap();
     }