Fix UB when interacting with Stream::Null(_)

src/lib/builtins.pl

@@ -2188,12 +2188,10 @@
 %% at_end_of_stream(+Stream).
 %
 % True iff the stream Stream has ended
-at_end_of_stream(S_or_a) :-
-    (  var(S_or_a) ->
+at_end_of_stream(S) :-
+    (  var(S) ->
        throw(error(instantiation_error, at_end_of_stream/1))
-    ;  atom(S_or_a) ->
-       stream_property(S, alias(S_or_a))
-    ;  S = S_or_a
+    ;  true
     ),
     stream_property(S, end_of_stream(E)),
     ( E = at -> true ; E = past ).
@@ -2205,7 +2203,7 @@
     current_input(S),
     stream_property(S, end_of_stream(E)),
     !,
-    ( E = at ; E = past ).
+    ( E = at -> true ; E = past ).
 
 %% set_stream_position(+Stream, +Position).
 %

src/machine/lib_machine/mod.rs

@@ -535,7 +535,7 @@ impl Machine {
     /// Consults a module into the [`Machine`] from a string.
     pub fn consult_module_string(&mut self, module_name: &str, program: impl Into<String>) {
         let stream = Stream::from_owned_string(program.into(), &mut self.machine_st.arena);
-        self.machine_st.registers[1] = stream_as_cell!(stream);
+        self.machine_st.registers[1] = stream.into();
         self.machine_st.registers[2] = atom_as_cell!(&atom_table::AtomTable::build_with(
             &self.machine_st.atom_tbl,
             module_name

src/machine/mod.rs

@@ -291,7 +291,7 @@ impl Machine {
     }
 
     fn load_file(&mut self, path: &str, stream: Stream) {
-        self.machine_st.registers[1] = stream_as_cell!(stream);
+        self.machine_st.registers[1] = stream.into();
         self.machine_st.registers[2] =
             atom_as_cell!(AtomTable::build_with(&self.machine_st.atom_tbl, path));
 
@@ -509,6 +509,13 @@ impl Machine {
             .insert(atom!("user_error"), self.user_error);
 
         self.indices.streams.insert(self.user_error);
+
+        let mut null_options = StreamOptions::default();
+        null_options.set_alias_to_atom_opt(Some(atom!("null_stream")));
+
+        self.indices
+            .stream_aliases
+            .insert(atom!("null_stream"), Stream::Null(null_options));
     }
 
     #[inline(always)]

src/machine/streams.rs

@@ -836,13 +836,13 @@ impl Read for Stream {
                 ErrorKind::PermissionDenied,
                 StreamError::ReadFromOutputStream,
             )),
-            Stream::OutputFile(_)
-            | Stream::StandardError(_)
-            | Stream::StandardOutput(_)
-            | Stream::Null(_) => Err(std::io::Error::new(
-                ErrorKind::PermissionDenied,
-                StreamError::ReadFromOutputStream,
-            )),
+            Stream::Null(_) => Ok(0),
+            Stream::OutputFile(_) | Stream::StandardError(_) | Stream::StandardOutput(_) => {
+                Err(std::io::Error::new(
+                    ErrorKind::PermissionDenied,
+                    StreamError::ReadFromOutputStream,
+                ))
+            }
         }
     }
 }
@@ -864,13 +864,10 @@ impl Write for Stream {
                 ErrorKind::PermissionDenied,
                 StreamError::WriteToInputStream,
             )),
-            Stream::StaticString(_)
-            | Stream::Readline(_)
-            | Stream::InputFile(..)
-            | Stream::Null(_) => Err(std::io::Error::new(
-                ErrorKind::PermissionDenied,
-                StreamError::WriteToInputStream,
-            )),
+            Stream::Null(_) => Ok(buf.len()),
+            Stream::StaticString(_) | Stream::Readline(_) | Stream::InputFile(..) => Err(
+                std::io::Error::new(ErrorKind::PermissionDenied, StreamError::WriteToInputStream),
+            ),
         }
     }
 
@@ -890,13 +887,10 @@ impl Write for Stream {
                 ErrorKind::PermissionDenied,
                 StreamError::FlushToInputStream,
             )),
-            Stream::StaticString(_)
-            | Stream::Readline(_)
-            | Stream::InputFile(_)
-            | Stream::Null(_) => Err(std::io::Error::new(
-                ErrorKind::PermissionDenied,
-                StreamError::FlushToInputStream,
-            )),
+            Stream::Null(_) => Ok(()),
+            Stream::StaticString(_) | Stream::Readline(_) | Stream::InputFile(_) => Err(
+                std::io::Error::new(ErrorKind::PermissionDenied, StreamError::FlushToInputStream),
+            ),
         }
     }
 }
@@ -982,6 +976,20 @@ fn cursor_position<T>(
     }
 }
 
+impl From<Stream> for HeapCellValue {
+    #[inline(always)]
+    fn from(stream: Stream) -> Self {
+        if stream.is_null_stream() {
+            let res = atom!("null_stream");
+            atom_as_cell!(res)
+        } else {
+            let res = stream.as_ptr();
+            debug_assert!(!res.is_null());
+            raw_ptr_as_cell!(res)
+        }
+    }
+}
+
 impl Stream {
     #[inline]
     pub(crate) fn position(&mut self) -> Option<(u64, usize)> {
@@ -1130,6 +1138,7 @@ impl Stream {
                     }
                 }
             }
+            Stream::Null(_) => AtEndOfStream::At,
             #[cfg(feature = "http")]
             Stream::HttpRead(stream_layout) => {
                 if stream_layout
@@ -1342,7 +1351,8 @@ impl Stream {
             | Stream::Byte(_)
             | Stream::Readline(_)
             | Stream::StaticString(_)
-            | Stream::InputFile(..) => true,
+            | Stream::InputFile(..)
+            | Stream::Null(_) => true,
             _ => false,
         }
     }
@@ -1358,7 +1368,8 @@ impl Stream {
             | Stream::StandardOutput(_)
             | Stream::NamedTcp(..)
             | Stream::Byte(_)
-            | Stream::OutputFile(..) => true,
+            | Stream::OutputFile(..)
+            | Stream::Null(_) => true,
             _ => false,
         }
     }
@@ -1593,7 +1604,7 @@ impl MachineState {
                             debug_assert_eq!(arity, 0);
 
                             return match stream_aliases.get(&name) {
-                                Some(stream) if !stream.is_null_stream() => Ok(*stream),
+                                Some(stream) => Ok(*stream),
                                 _ => {
                                     let stub = functor_stub(caller, arity);
                                     let addr = atom_as_cell!(name);
@@ -1611,7 +1622,7 @@ impl MachineState {
                             debug_assert_eq!(arity, 0);
 
                             return match stream_aliases.get(&name) {
-                                Some(stream) if !stream.is_null_stream() => Ok(*stream),
+                                Some(stream) => Ok(*stream),
                                 _ => {
                                     let stub = functor_stub(caller, arity);
                                     let addr = atom_as_cell!(name);
@@ -1625,11 +1636,10 @@ impl MachineState {
                         (HeapCellValueTag::Cons, ptr) => {
                             match_untyped_arena_ptr!(ptr,
                                 (ArenaHeaderTag::Stream, stream) => {
-                                    return if stream.is_null_stream() {
-                                        Err(self.open_permission_error(stream_as_cell!(stream), caller, arity))
-                                    } else {
-                                        Ok(stream)
-                                    };
+                                    if stream.is_null_stream() {
+                                        unreachable!("Null streams have no Cons representation");
+                                    }
+                                    return Ok(stream);
                                 }
                                 (ArenaHeaderTag::Dropped, _value) => {
                                     let stub = functor_stub(caller, arity);
@@ -1689,7 +1699,7 @@ impl MachineState {
             if let Some(alias) = stream.options().get_alias() {
                 atom_as_cell!(alias)
             } else {
-                stream_as_cell!(stream)
+                stream.into()
             },
         );
 
@@ -1893,3 +1903,158 @@ impl MachineState {
         }
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::machine::config::*;
+    use crate::LeafAnswer;
+
+    fn is_successful<T>(answer: &Result<LeafAnswer, T>) -> bool {
+        matches!(
+            answer,
+            Ok(LeafAnswer::True) | Ok(LeafAnswer::LeafAnswer { .. })
+        )
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn current_input_null_stream() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine.run_query("current_input(S).").collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(is_successful(&results[0]));
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn read_null_stream() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine.run_query("get_code(C).").collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(
+            is_successful(&results[0]),
+            "Expected read to succeed, got {:?}",
+            results[0]
+        );
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn current_output_null_stream() {
+        // TODO: switch to a proper solution for configuring the machine with null streams
+        // once `StreamConfig` supports it.
+        let mut machine = MachineBuilder::new().build();
+        machine.user_output = Stream::Null(StreamOptions::default());
+        machine.configure_streams();
+
+        let results = machine.run_query("current_output(S).").collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(is_successful(&results[0]));
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn write_null_stream() {
+        // TODO: switch to a proper solution for configuring the machine with null streams
+        // once `StreamConfig` supports it.
+        let mut machine = MachineBuilder::new().build();
+        machine.user_output = Stream::Null(StreamOptions::default());
+        machine.configure_streams();
+
+        let results = machine.run_query("write(hello).").collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(
+            is_successful(&results[0]),
+            "Expected write to succeed, got {:?}",
+            results[0]
+        );
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn put_code_null_stream() {
+        // TODO: switch to a proper solution for configuring the machine with null streams
+        // once `StreamConfig` supports it.
+        let mut machine = MachineBuilder::new().build();
+        machine.user_output = Stream::Null(StreamOptions::default());
+        machine.configure_streams();
+
+        let results = machine
+            .run_query("put_code(user_output, 65).")
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(
+            is_successful(&results[0]),
+            "Expected write to succeed, got {:?}",
+            results[0]
+        );
+    }
+
+    /// A variant of the [`write_null_stream`] that tries to write to a (null) input stream.
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn write_null_input_stream() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine
+            .run_query("current_input(Stream), write(Stream, hello).")
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(
+            is_successful(&results[0]),
+            "Expected write to succeed, got {:?}",
+            results[0]
+        );
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn at_end_of_stream_0_null_stream() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine.run_query("at_end_of_stream.").collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(
+            is_successful(&results[0]),
+            "Expected at_end_of_stream to succeed, got {:?}",
+            results[0]
+        );
+    }
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn at_end_of_stream_1_null_stream() {
+        let mut machine = MachineBuilder::new()
+            .with_streams(StreamConfig::in_memory())
+            .build();
+
+        let results = machine
+            .run_query("current_input(Stream), at_end_of_stream(Stream).")
+            .collect::<Vec<_>>();
+
+        assert_eq!(results.len(), 1);
+        assert!(
+            is_successful(&results[0]),
+            "Expected at_end_of_stream to succeed, got {:?}",
+            results[0]
+        );
+    }
+}