diff --git a/Cargo.lock b/Cargo.lock
index 2b754e8af..9ed5ce894 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7194,6 +7194,7 @@ dependencies = [
 name = "reth-engine-primitives"
 version = "1.1.2"
 dependencies = [
+ "alloy-consensus",
  "alloy-primitives",
  "alloy-rpc-types-engine",
  "futures",
@@ -7202,6 +7203,7 @@ dependencies = [
  "reth-payload-builder-primitives",
  "reth-payload-primitives",
  "reth-primitives",
+ "reth-primitives-traits",
  "reth-trie",
  "serde",
  "thiserror 1.0.69",
@@ -7431,6 +7433,7 @@ dependencies = [
  "reth-chainspec",
  "reth-engine-primitives",
  "reth-payload-primitives",
+ "reth-payload-validator",
  "reth-primitives",
  "reth-rpc-types-compat",
  "serde",
@@ -8348,6 +8351,7 @@ dependencies = [
  "reth-optimism-rpc",
  "reth-payload-builder",
  "reth-payload-util",
+ "reth-payload-validator",
  "reth-primitives",
  "reth-provider",
  "reth-revm",
diff --git a/crates/engine/primitives/Cargo.toml b/crates/engine/primitives/Cargo.toml
index 42cbd932d..2da1be9c9 100644
--- a/crates/engine/primitives/Cargo.toml
+++ b/crates/engine/primitives/Cargo.toml
@@ -16,11 +16,13 @@ reth-execution-types.workspace = true
 reth-payload-primitives.workspace = true
 reth-payload-builder-primitives.workspace = true
 reth-primitives.workspace = true
+reth-primitives-traits.workspace = true
 reth-trie.workspace = true
 reth-errors.workspace = true
 
 # alloy
 alloy-primitives.workspace = true
+alloy-consensus.workspace = true
 alloy-rpc-types-engine.workspace = true
 
 # async
diff --git a/crates/engine/primitives/src/lib.rs b/crates/engine/primitives/src/lib.rs
index 3429edc28..89fb7459b 100644
--- a/crates/engine/primitives/src/lib.rs
+++ b/crates/engine/primitives/src/lib.rs
@@ -9,6 +9,9 @@
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
 mod error;
+
+use alloy_consensus::BlockHeader;
+use alloy_rpc_types_engine::{ExecutionPayload, ExecutionPayloadSidecar, PayloadError};
 pub use error::BeaconOnNewPayloadError;
 
 mod forkchoice;
@@ -24,6 +27,9 @@ pub use reth_payload_primitives::{
     BuiltPayload, EngineApiMessageVersion, EngineObjectValidationError, PayloadOrAttributes,
     PayloadTypes,
 };
+use reth_payload_primitives::{InvalidPayloadAttributesError, PayloadAttributes};
+use reth_primitives::SealedBlockFor;
+use reth_primitives_traits::Block;
 use serde::{de::DeserializeOwned, ser::Serialize};
 
 /// This type defines the versioned types of the engine API.
@@ -74,8 +80,11 @@ pub trait EngineTypes:
         + 'static;
 }
 
-/// Type that validates the payloads sent to the engine.
+/// Type that validates the payloads processed by the engine.
 pub trait EngineValidator<Types: EngineTypes>: Clone + Send + Sync + Unpin + 'static {
+    /// The block type used by the engine.
+    type Block: Block;
+
     /// Validates the presence or exclusion of fork-specific fields based on the payload attributes
     /// and the message version.
     fn validate_version_specific_fields(
@@ -90,4 +99,38 @@ pub trait EngineValidator<Types: EngineTypes>: Clone + Send + Sync + Unpin + 'st
         version: EngineApiMessageVersion,
         attributes: &<Types as PayloadTypes>::PayloadAttributes,
     ) -> Result<(), EngineObjectValidationError>;
+
+    /// Ensures that the given payload does not violate any consensus rules that concern the block's
+    /// layout.
+    ///
+    /// This function must convert the payload into the executable block and pre-validate its
+    /// fields.
+    ///
+    /// Implementers should ensure that the checks are done in the order that conforms with the
+    /// engine-API specification.
+    fn ensure_well_formed_payload(
+        &self,
+        payload: ExecutionPayload,
+        sidecar: ExecutionPayloadSidecar,
+    ) -> Result<SealedBlockFor<Self::Block>, PayloadError>;
+
+    /// Validates the payload attributes with respect to the header.
+    ///
+    /// By default, this enforces that the payload attributes timestamp is greater than the
+    /// timestamp according to:
+    ///   > 7. Client software MUST ensure that payloadAttributes.timestamp is greater than
+    ///   > timestamp
+    ///   > of a block referenced by forkchoiceState.headBlockHash.
+    ///
+    /// See also [engine api spec](https://github.com/ethereum/execution-apis/tree/fe8e13c288c592ec154ce25c534e26cb7ce0530d/src/engine)
+    fn validate_payload_attributes_against_header(
+        &self,
+        attr: &<Types as PayloadTypes>::PayloadAttributes,
+        header: &<Self::Block as Block>::Header,
+    ) -> Result<(), InvalidPayloadAttributesError> {
+        if attr.timestamp() <= header.timestamp() {
+            return Err(InvalidPayloadAttributesError::InvalidTimestamp);
+        }
+        Ok(())
+    }
 }
diff --git a/crates/ethereum/engine-primitives/Cargo.toml b/crates/ethereum/engine-primitives/Cargo.toml
index e9bcd4256..f019f6e5f 100644
--- a/crates/ethereum/engine-primitives/Cargo.toml
+++ b/crates/ethereum/engine-primitives/Cargo.toml
@@ -16,6 +16,7 @@ reth-chainspec.workspace = true
 reth-primitives.workspace = true
 reth-engine-primitives.workspace = true
 reth-payload-primitives.workspace = true
+reth-payload-validator.workspace = true
 reth-rpc-types-compat.workspace = true
 alloy-rlp.workspace = true
 reth-chain-state.workspace = true
diff --git a/crates/ethereum/engine-primitives/src/lib.rs b/crates/ethereum/engine-primitives/src/lib.rs
index 5addf2a18..beefd54ca 100644
--- a/crates/ethereum/engine-primitives/src/lib.rs
+++ b/crates/ethereum/engine-primitives/src/lib.rs
@@ -11,6 +11,7 @@
 mod payload;
 use std::sync::Arc;
 
+use alloy_rpc_types_engine::{ExecutionPayload, ExecutionPayloadSidecar, PayloadError};
 pub use alloy_rpc_types_engine::{
     ExecutionPayloadEnvelopeV2, ExecutionPayloadEnvelopeV3, ExecutionPayloadEnvelopeV4,
     ExecutionPayloadV1, PayloadAttributes as EthPayloadAttributes,
@@ -22,6 +23,8 @@ use reth_payload_primitives::{
     validate_version_specific_fields, EngineApiMessageVersion, EngineObjectValidationError,
     PayloadOrAttributes, PayloadTypes,
 };
+use reth_payload_validator::ExecutionPayloadValidator;
+use reth_primitives::{Block, SealedBlock};
 
 /// The types used in the default mainnet ethereum beacon consensus engine.
 #[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
@@ -63,13 +66,19 @@ impl PayloadTypes for EthPayloadTypes {
 /// Validator for the ethereum engine API.
 #[derive(Debug, Clone)]
 pub struct EthereumEngineValidator {
-    chain_spec: Arc<ChainSpec>,
+    inner: ExecutionPayloadValidator<ChainSpec>,
 }
 
 impl EthereumEngineValidator {
     /// Instantiates a new validator.
     pub const fn new(chain_spec: Arc<ChainSpec>) -> Self {
-        Self { chain_spec }
+        Self { inner: ExecutionPayloadValidator::new(chain_spec) }
+    }
+
+    /// Returns the chain spec used by the validator.
+    #[inline]
+    fn chain_spec(&self) -> &ChainSpec {
+        self.inner.chain_spec()
     }
 }
 
@@ -77,12 +86,14 @@ impl<Types> EngineValidator<Types> for EthereumEngineValidator
 where
     Types: EngineTypes<PayloadAttributes = EthPayloadAttributes>,
 {
+    type Block = Block;
+
     fn validate_version_specific_fields(
         &self,
         version: EngineApiMessageVersion,
         payload_or_attrs: PayloadOrAttributes<'_, EthPayloadAttributes>,
     ) -> Result<(), EngineObjectValidationError> {
-        validate_version_specific_fields(&self.chain_spec, version, payload_or_attrs)
+        validate_version_specific_fields(self.chain_spec(), version, payload_or_attrs)
     }
 
     fn ensure_well_formed_attributes(
@@ -90,6 +101,14 @@ where
         version: EngineApiMessageVersion,
         attributes: &EthPayloadAttributes,
     ) -> Result<(), EngineObjectValidationError> {
-        validate_version_specific_fields(&self.chain_spec, version, attributes.into())
+        validate_version_specific_fields(self.chain_spec(), version, attributes.into())
+    }
+
+    fn ensure_well_formed_payload(
+        &self,
+        payload: ExecutionPayload,
+        sidecar: ExecutionPayloadSidecar,
+    ) -> Result<SealedBlock, PayloadError> {
+        self.inner.ensure_well_formed_payload(payload, sidecar)
     }
 }
diff --git a/crates/optimism/node/Cargo.toml b/crates/optimism/node/Cargo.toml
index fbc055a82..f11eb3578 100644
--- a/crates/optimism/node/Cargo.toml
+++ b/crates/optimism/node/Cargo.toml
@@ -18,6 +18,7 @@ reth-engine-local.workspace = true
 reth-primitives.workspace = true
 reth-payload-builder.workspace = true
 reth-payload-util.workspace = true
+reth-payload-validator.workspace = true
 reth-basic-payload-builder.workspace = true
 reth-consensus.workspace = true
 reth-node-api.workspace = true
diff --git a/crates/optimism/node/src/engine.rs b/crates/optimism/node/src/engine.rs
index dd4d0c13f..57b76b904 100644
--- a/crates/optimism/node/src/engine.rs
+++ b/crates/optimism/node/src/engine.rs
@@ -1,6 +1,7 @@
-use std::sync::Arc;
-
-use alloy_rpc_types_engine::{ExecutionPayloadEnvelopeV2, ExecutionPayloadV1};
+use alloy_rpc_types_engine::{
+    ExecutionPayload, ExecutionPayloadEnvelopeV2, ExecutionPayloadSidecar, ExecutionPayloadV1,
+    PayloadError,
+};
 use op_alloy_rpc_types_engine::{
     OpExecutionPayloadEnvelopeV3, OpExecutionPayloadEnvelopeV4, OpPayloadAttributes,
 };
@@ -16,6 +17,9 @@ use reth_node_api::{
 use reth_optimism_chainspec::OpChainSpec;
 use reth_optimism_forks::{OpHardfork, OpHardforks};
 use reth_optimism_payload_builder::{OpBuiltPayload, OpPayloadBuilderAttributes};
+use reth_payload_validator::ExecutionPayloadValidator;
+use reth_primitives::{Block, SealedBlockFor};
+use std::sync::Arc;
 
 /// The types used in the optimism beacon consensus engine.
 #[derive(Debug, Default, Clone, serde::Deserialize, serde::Serialize)]
@@ -57,13 +61,88 @@ impl PayloadTypes for OpPayloadTypes {
 /// Validator for Optimism engine API.
 #[derive(Debug, Clone)]
 pub struct OpEngineValidator {
-    chain_spec: Arc<OpChainSpec>,
+    inner: ExecutionPayloadValidator<OpChainSpec>,
 }
 
 impl OpEngineValidator {
     /// Instantiates a new validator.
     pub const fn new(chain_spec: Arc<OpChainSpec>) -> Self {
-        Self { chain_spec }
+        Self { inner: ExecutionPayloadValidator::new(chain_spec) }
+    }
+
+    /// Returns the chain spec used by the validator.
+    #[inline]
+    fn chain_spec(&self) -> &OpChainSpec {
+        self.inner.chain_spec()
+    }
+}
+
+impl<Types> EngineValidator<Types> for OpEngineValidator
+where
+    Types: EngineTypes<PayloadAttributes = OpPayloadAttributes>,
+{
+    type Block = Block;
+
+    fn validate_version_specific_fields(
+        &self,
+        version: EngineApiMessageVersion,
+        payload_or_attrs: PayloadOrAttributes<'_, OpPayloadAttributes>,
+    ) -> Result<(), EngineObjectValidationError> {
+        validate_withdrawals_presence(
+            self.chain_spec(),
+            version,
+            payload_or_attrs.message_validation_kind(),
+            payload_or_attrs.timestamp(),
+            payload_or_attrs.withdrawals().is_some(),
+        )?;
+        validate_parent_beacon_block_root_presence(
+            self.chain_spec(),
+            version,
+            payload_or_attrs.message_validation_kind(),
+            payload_or_attrs.timestamp(),
+            payload_or_attrs.parent_beacon_block_root().is_some(),
+        )
+    }
+
+    fn ensure_well_formed_attributes(
+        &self,
+        version: EngineApiMessageVersion,
+        attributes: &OpPayloadAttributes,
+    ) -> Result<(), EngineObjectValidationError> {
+        validate_version_specific_fields(self.chain_spec(), version, attributes.into())?;
+
+        if attributes.gas_limit.is_none() {
+            return Err(EngineObjectValidationError::InvalidParams(
+                "MissingGasLimitInPayloadAttributes".to_string().into(),
+            ))
+        }
+
+        if self
+            .chain_spec()
+            .is_holocene_active_at_timestamp(attributes.payload_attributes.timestamp)
+        {
+            let (elasticity, denominator) =
+                attributes.decode_eip_1559_params().ok_or_else(|| {
+                    EngineObjectValidationError::InvalidParams(
+                        "MissingEip1559ParamsInPayloadAttributes".to_string().into(),
+                    )
+                })?;
+            if elasticity != 0 && denominator == 0 {
+                return Err(EngineObjectValidationError::InvalidParams(
+                    "Eip1559ParamsDenominatorZero".to_string().into(),
+                ))
+            }
+        }
+
+        Ok(())
+    }
+
+    fn ensure_well_formed_payload(
+        &self,
+        payload: ExecutionPayload,
+        sidecar: ExecutionPayloadSidecar,
+    ) -> Result<SealedBlockFor<Self::Block>, PayloadError> {
+        self.inner.ensure_well_formed_payload(payload, sidecar)
     }
 }
 
@@ -109,63 +188,6 @@ pub fn validate_withdrawals_presence(
     Ok(())
 }
 
-impl<Types> EngineValidator<Types> for OpEngineValidator
-where
-    Types: EngineTypes<PayloadAttributes = OpPayloadAttributes>,
-{
-    fn validate_version_specific_fields(
-        &self,
-        version: EngineApiMessageVersion,
-        payload_or_attrs: PayloadOrAttributes<'_, OpPayloadAttributes>,
-    ) -> Result<(), EngineObjectValidationError> {
-        validate_withdrawals_presence(
-            &self.chain_spec,
-            version,
-            payload_or_attrs.message_validation_kind(),
-            payload_or_attrs.timestamp(),
-            payload_or_attrs.withdrawals().is_some(),
-        )?;
-        validate_parent_beacon_block_root_presence(
-            &self.chain_spec,
-            version,
-            payload_or_attrs.message_validation_kind(),
-            payload_or_attrs.timestamp(),
-            payload_or_attrs.parent_beacon_block_root().is_some(),
-        )
-    }
-
-    fn ensure_well_formed_attributes(
-        &self,
-        version: EngineApiMessageVersion,
-        attributes: &OpPayloadAttributes,
-    ) -> Result<(), EngineObjectValidationError> {
-        validate_version_specific_fields(&self.chain_spec, version, attributes.into())?;
-
-        if attributes.gas_limit.is_none() {
-            return Err(EngineObjectValidationError::InvalidParams(
-                "MissingGasLimitInPayloadAttributes".to_string().into(),
-            ))
-        }
-
-        if self.chain_spec.is_holocene_active_at_timestamp(attributes.payload_attributes.timestamp)
-        {
-            let (elasticity, denominator) =
-                attributes.decode_eip_1559_params().ok_or_else(|| {
-                    EngineObjectValidationError::InvalidParams(
-                        "MissingEip1559ParamsInPayloadAttributes".to_string().into(),
-                    )
-                })?;
-            if elasticity != 0 && denominator == 0 {
-                return Err(EngineObjectValidationError::InvalidParams(
-                    "Eip1559ParamsDenominatorZero".to_string().into(),
-                ))
-            }
-        }
-
-        Ok(())
-    }
-}
-
 #[cfg(test)]
 mod test {
 
diff --git a/crates/payload/primitives/src/error.rs b/crates/payload/primitives/src/error.rs
index d2e57da57..ffe4e027e 100644
--- a/crates/payload/primitives/src/error.rs
+++ b/crates/payload/primitives/src/error.rs
@@ -1,6 +1,7 @@
 //! Error types emitted by types or implementations of this crate.
 
 use alloy_primitives::B256;
+use alloy_rpc_types_engine::ForkchoiceUpdateError;
 use reth_errors::{ProviderError, RethError};
 use revm_primitives::EVMError;
 use tokio::sync::oneshot;
@@ -53,7 +54,7 @@ impl From<oneshot::error::RecvError> for PayloadBuilderError {
     }
 }
 
-/// Thrown when the payload or attributes are known to be invalid before processing.
+/// Thrown when the payload or attributes are known to be invalid __before__ processing.
 ///
 /// This is used mainly for
 /// [`validate_version_specific_fields`](crate::validate_version_specific_fields), which validates
@@ -115,3 +116,20 @@ impl EngineObjectValidationError {
         Self::InvalidParams(Box::new(error))
     }
 }
+
+/// Thrown when validating the correctness of a payloadattributes object.
+#[derive(thiserror::Error, Debug)]
+pub enum InvalidPayloadAttributesError {
+    /// Thrown if the timestamp of the payload attributes is invalid according to the engine specs.
+    #[error("parent beacon block root not supported before V3")]
+    InvalidTimestamp,
+    /// Another type of error that is not covered by the above variants.
+    #[error("Invalid params: {0}")]
+    InvalidParams(#[from] Box<dyn core::error::Error + Send + Sync>),
+}
+
+impl From<InvalidPayloadAttributesError> for ForkchoiceUpdateError {
+    fn from(_: InvalidPayloadAttributesError) -> Self {
+        Self::UpdatedInvalidPayloadAttributes
+    }
+}
diff --git a/crates/payload/primitives/src/lib.rs b/crates/payload/primitives/src/lib.rs
index 0ff4810b8..523e6fb05 100644
--- a/crates/payload/primitives/src/lib.rs
+++ b/crates/payload/primitives/src/lib.rs
@@ -9,7 +9,10 @@
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
 mod error;
-pub use error::{EngineObjectValidationError, PayloadBuilderError, VersionSpecificValidationError};
+pub use error::{
+    EngineObjectValidationError, InvalidPayloadAttributesError, PayloadBuilderError,
+    VersionSpecificValidationError,
+};
 
 /// Contains traits to abstract over payload attributes types and default implementations of the
 /// [`PayloadAttributes`] trait for ethereum mainnet and optimism types.
diff --git a/examples/custom-engine-types/src/main.rs b/examples/custom-engine-types/src/main.rs
index cde68ca6d..f9ac5c238 100644
--- a/examples/custom-engine-types/src/main.rs
+++ b/examples/custom-engine-types/src/main.rs
@@ -17,11 +17,6 @@
 
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 
-use std::{convert::Infallible, sync::Arc};
-
-use serde::{Deserialize, Serialize};
-use thiserror::Error;
-
 use alloy_eips::eip4895::Withdrawals;
 use alloy_genesis::Genesis;
 use alloy_primitives::{Address, B256};
@@ -33,7 +28,7 @@ use alloy_rpc_types::{
     Withdrawal,
 };
 use reth::{
-    api::PayloadTypes,
+    api::{InvalidPayloadAttributesError, PayloadTypes},
     builder::{
         components::{ComponentsBuilder, PayloadServiceBuilder},
         node::{NodeTypes, NodeTypesWithEngine},
@@ -42,9 +37,13 @@ use reth::{
         PayloadBuilderConfig,
     },
     network::NetworkHandle,
-    primitives::EthPrimitives,
+    payload::ExecutionPayloadValidator,
+    primitives::{Block, EthPrimitives, SealedBlockFor},
     providers::{CanonStateSubscriptions, EthStorage, StateProviderFactory},
-    rpc::eth::EthApi,
+    rpc::{
+        eth::EthApi,
+        types::engine::{ExecutionPayload, ExecutionPayloadSidecar, PayloadError},
+    },
     tasks::TaskManager,
     transaction_pool::TransactionPool,
 };
@@ -72,6 +71,9 @@ use reth_payload_builder::{
 };
 use reth_tracing::{RethTracer, Tracer};
 use reth_trie_db::MerklePatriciaTrie;
+use serde::{Deserialize, Serialize};
+use std::{convert::Infallible, sync::Arc};
+use thiserror::Error;
 
 /// A custom payload attributes type.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -171,19 +173,34 @@ impl EngineTypes for CustomEngineTypes {
 /// Custom engine validator
 #[derive(Debug, Clone)]
 pub struct CustomEngineValidator {
-    chain_spec: Arc<ChainSpec>,
+    inner: ExecutionPayloadValidator<ChainSpec>,
+}
+
+impl CustomEngineValidator {
+    /// Instantiates a new validator.
+    pub const fn new(chain_spec: Arc<ChainSpec>) -> Self {
+        Self { inner: ExecutionPayloadValidator::new(chain_spec) }
+    }
+
+    /// Returns the chain spec used by the validator.
+    #[inline]
+    fn chain_spec(&self) -> &ChainSpec {
+        self.inner.chain_spec()
+    }
 }
 
 impl<T> EngineValidator<T> for CustomEngineValidator
 where
     T: EngineTypes<PayloadAttributes = CustomPayloadAttributes>,
 {
+    type Block = Block;
+
     fn validate_version_specific_fields(
         &self,
         version: EngineApiMessageVersion,
         payload_or_attrs: PayloadOrAttributes<'_, T::PayloadAttributes>,
     ) -> Result<(), EngineObjectValidationError> {
-        validate_version_specific_fields(&self.chain_spec, version, payload_or_attrs)
+        validate_version_specific_fields(self.chain_spec(), version, payload_or_attrs)
     }
 
     fn ensure_well_formed_attributes(
@@ -191,7 +208,7 @@ where
         version: EngineApiMessageVersion,
         attributes: &T::PayloadAttributes,
     ) -> Result<(), EngineObjectValidationError> {
-        validate_version_specific_fields(&self.chain_spec, version, attributes.into())?;
+        validate_version_specific_fields(self.chain_spec(), version, attributes.into())?;
 
         // custom validation logic - ensure that the custom field is not zero
         if attributes.custom == 0 {
@@ -202,6 +219,23 @@ where
 
         Ok(())
     }
+
+    fn ensure_well_formed_payload(
+        &self,
+        payload: ExecutionPayload,
+        sidecar: ExecutionPayloadSidecar,
+    ) -> Result<SealedBlockFor<Self::Block>, PayloadError> {
+        self.inner.ensure_well_formed_payload(payload, sidecar)
+    }
+
+    fn validate_payload_attributes_against_header(
+        &self,
+        _attr: &<T as PayloadTypes>::PayloadAttributes,
+        _header: &<Self::Block as reth::api::Block>::Header,
+    ) -> Result<(), InvalidPayloadAttributesError> {
+        // skip default timestamp validation
+        Ok(())
+    }
 }
 
 /// Custom engine validator builder
@@ -218,7 +252,7 @@ where
     type Validator = CustomEngineValidator;
 
     async fn build(self, ctx: &AddOnsContext<'_, N>) -> eyre::Result<Self::Validator> {
-        Ok(CustomEngineValidator { chain_spec: ctx.config.chain.clone() })
+        Ok(CustomEngineValidator::new(ctx.config.chain.clone()))
     }
 }