package sealing import ( "bytes" "context" "net/url" "github.com/ipfs/go-cid" "github.com/ipfs/go-datastore" "github.com/multiformats/go-multihash" "golang.org/x/xerrors" "github.com/filecoin-project/go-address" "github.com/filecoin-project/go-state-types/abi" "github.com/filecoin-project/go-state-types/crypto" "github.com/filecoin-project/go-state-types/proof" "github.com/filecoin-project/go-statemachine" "github.com/filecoin-project/lotus/api" "github.com/filecoin-project/lotus/storage/sealer/storiface" ) func (m *Sealing) Receive(ctx context.Context, meta api.RemoteSectorMeta) error { m.inputLk.Lock() defer m.inputLk.Unlock() si, err := m.checkSectorMeta(ctx, meta) if err != nil { return err } exists, err := m.sectors.Has(uint64(meta.Sector.Number)) if err != nil { return xerrors.Errorf("checking if sector exists: %w", err) } if exists { return xerrors.Errorf("sector %d state already exists", meta.Sector.Number) } err = m.sectors.Send(uint64(meta.Sector.Number), SectorReceive{ State: si, }) if err != nil { return xerrors.Errorf("receiving sector: %w", err) } return nil } func (m *Sealing) checkSectorMeta(ctx context.Context, meta api.RemoteSectorMeta) (SectorInfo, error) { { mid, err := address.IDFromAddress(m.maddr) if err != nil { panic(err) } if meta.Sector.Miner != abi.ActorID(mid) { return SectorInfo{}, xerrors.Errorf("sector for wrong actor - expected actor id %d, sector was for actor %d", mid, meta.Sector.Miner) } } { // initial sanity check, doesn't prevent races _, err := m.GetSectorInfo(meta.Sector.Number) if err != nil && !xerrors.Is(err, datastore.ErrNotFound) { return SectorInfo{}, err } if err == nil { return SectorInfo{}, xerrors.Errorf("sector with ID %d already exists in the sealing pipeline", meta.Sector.Number) } } { spt, err := m.currentSealProof(ctx) if err != nil { return SectorInfo{}, err } if meta.Type != spt { return SectorInfo{}, xerrors.Errorf("sector seal proof type doesn't match current seal proof type (%d!=%d)", meta.Type, spt) } } ts, err := m.Api.ChainHead(ctx) if err != nil { return SectorInfo{}, xerrors.Errorf("getting chain head: %w", err) } var info SectorInfo var validatePoRep bool switch SectorState(meta.State) { case Proving, Available: if meta.CommitMessage != nil { if err := checkMessagePrefix(*meta.CommitMessage); err != nil { return SectorInfo{}, xerrors.Errorf("commit message prefix: %w", err) } info.CommitMessage = meta.CommitMessage } fallthrough case SubmitCommit: if meta.PreCommitDeposit == nil { return SectorInfo{}, xerrors.Errorf("sector PreCommitDeposit was null") } info.PreCommitDeposit = *meta.PreCommitDeposit info.PreCommitTipSet = meta.PreCommitTipSet if info.PreCommitMessage != nil { if err := checkMessagePrefix(*meta.PreCommitMessage); err != nil { return SectorInfo{}, xerrors.Errorf("commit message prefix: %w", err) } info.PreCommitMessage = meta.PreCommitMessage } // check provided seed if len(meta.SeedValue) != abi.RandomnessLength { return SectorInfo{}, xerrors.Errorf("seed randomness had wrong length %d", len(meta.SeedValue)) } maddrBuf := new(bytes.Buffer) if err := m.maddr.MarshalCBOR(maddrBuf); err != nil { return SectorInfo{}, xerrors.Errorf("marshal miner address for seed check: %w", err) } rand, err := m.Api.StateGetRandomnessFromTickets(ctx, crypto.DomainSeparationTag_InteractiveSealChallengeSeed, meta.SeedEpoch, maddrBuf.Bytes(), ts.Key()) if err != nil { return SectorInfo{}, xerrors.Errorf("generating check seed: %w", err) } if !bytes.Equal(rand, meta.SeedValue) { return SectorInfo{}, xerrors.Errorf("provided(%x) and generated(%x) seeds differ", meta.SeedValue, rand) } info.SeedValue = meta.SeedValue info.SeedEpoch = meta.SeedEpoch info.Proof = meta.CommitProof validatePoRep = true fallthrough case PreCommitting: // check provided ticket if len(meta.TicketValue) != abi.RandomnessLength { return SectorInfo{}, xerrors.Errorf("ticket randomness had wrong length %d", len(meta.TicketValue)) } maddrBuf := new(bytes.Buffer) if err := m.maddr.MarshalCBOR(maddrBuf); err != nil { return SectorInfo{}, xerrors.Errorf("marshal miner address for ticket check: %w", err) } rand, err := m.Api.StateGetRandomnessFromTickets(ctx, crypto.DomainSeparationTag_SealRandomness, meta.TicketEpoch, maddrBuf.Bytes(), ts.Key()) if err != nil { return SectorInfo{}, xerrors.Errorf("generating check ticket: %w", err) } if !bytes.Equal(rand, meta.TicketValue) { return SectorInfo{}, xerrors.Errorf("provided(%x) and generated(%x) tickets differ", meta.TicketValue, rand) } info.TicketValue = meta.TicketValue info.TicketEpoch = meta.TicketEpoch info.PreCommit1Out = meta.PreCommit1Out // check CommD/R if meta.CommD == nil || meta.CommR == nil { return SectorInfo{}, xerrors.Errorf("both CommR/CommD cids need to be set for sectors in PreCommitting and later states") } dp := meta.CommD.Prefix() if dp.Version != 1 || dp.Codec != cid.FilCommitmentUnsealed || dp.MhType != multihash.SHA2_256_TRUNC254_PADDED || dp.MhLength != 32 { return SectorInfo{}, xerrors.Errorf("CommD cid has wrong prefix") } rp := meta.CommR.Prefix() if rp.Version != 1 || rp.Codec != cid.FilCommitmentSealed || rp.MhType != multihash.POSEIDON_BLS12_381_A1_FC1 || rp.MhLength != 32 { return SectorInfo{}, xerrors.Errorf("CommR cid has wrong prefix") } info.CommD = meta.CommD info.CommR = meta.CommR if meta.DataSealed == nil { return SectorInfo{}, xerrors.Errorf("expected DataSealed to be set") } if meta.DataCache == nil { return SectorInfo{}, xerrors.Errorf("expected DataCache to be set") } info.RemoteDataSealed = meta.DataSealed // todo make head requests to check? info.RemoteDataCache = meta.DataCache if meta.RemoteCommit1Endpoint != "" { // validate the url if _, err := url.Parse(meta.RemoteCommit1Endpoint); err != nil { return SectorInfo{}, xerrors.Errorf("parsing remote c1 endpoint url: %w", err) } info.RemoteCommit1Endpoint = meta.RemoteCommit1Endpoint } if meta.RemoteCommit2Endpoint != "" { // validate the url if _, err := url.Parse(meta.RemoteCommit2Endpoint); err != nil { return SectorInfo{}, xerrors.Errorf("parsing remote c2 endpoint url: %w", err) } info.RemoteCommit2Endpoint = meta.RemoteCommit2Endpoint } // If we get a sector after PC2, and remote C1 endpoint is set, assume that we're getting finalized sector data if info.RemoteCommit1Endpoint != "" { info.RemoteDataFinalized = true } fallthrough case GetTicket, Packing: info.Return = ReturnState(meta.State) info.State = ReceiveSector info.SectorNumber = meta.Sector.Number info.Pieces = meta.Pieces info.SectorType = meta.Type if meta.RemoteSealingDoneEndpoint != "" { // validate the url if _, err := url.Parse(meta.RemoteSealingDoneEndpoint); err != nil { return SectorInfo{}, xerrors.Errorf("parsing remote sealing-done endpoint url: %w", err) } info.RemoteSealingDoneEndpoint = meta.RemoteSealingDoneEndpoint } if err := checkPieces(ctx, m.maddr, meta.Sector.Number, meta.Pieces, m.Api, false); err != nil { return SectorInfo{}, xerrors.Errorf("checking pieces: %w", err) } if meta.DataUnsealed == nil { return SectorInfo{}, xerrors.Errorf("expected DataUnsealed to be set") } info.RemoteDataUnsealed = meta.DataUnsealed // some late checks which require previous checks if validatePoRep { ok, err := m.verif.VerifySeal(proof.SealVerifyInfo{ SealProof: meta.Type, SectorID: meta.Sector, DealIDs: nil, Randomness: meta.TicketValue, InteractiveRandomness: meta.SeedValue, Proof: meta.CommitProof, SealedCID: *meta.CommR, UnsealedCID: *meta.CommD, }) if err != nil { return SectorInfo{}, xerrors.Errorf("validating seal proof: %w", err) } if !ok { return SectorInfo{}, xerrors.Errorf("seal proof invalid") } } return info, nil default: return SectorInfo{}, xerrors.Errorf("imported sector State in not supported") } } func (m *Sealing) handleReceiveSector(ctx statemachine.Context, sector SectorInfo) error { toFetch := map[storiface.SectorFileType]storiface.SectorLocation{} for fileType, data := range map[storiface.SectorFileType]*storiface.SectorLocation{ storiface.FTUnsealed: sector.RemoteDataUnsealed, storiface.FTSealed: sector.RemoteDataSealed, storiface.FTCache: sector.RemoteDataCache, } { if data == nil { continue } if data.Local { // todo check exists continue } toFetch[fileType] = *data } if len(toFetch) > 0 { if err := m.sealer.DownloadSectorData(ctx.Context(), m.minerSector(sector.SectorType, sector.SectorNumber), sector.RemoteDataFinalized, toFetch); err != nil { return xerrors.Errorf("downloading sector data: %w", err) // todo send err event } } // todo data checks? return ctx.Send(SectorReceived{}) } func checkMessagePrefix(c cid.Cid) error { p := c.Prefix() if p.Version != 1 || p.MhLength != 32 || p.MhType != multihash.BLAKE2B_MIN+31 || p.Codec != cid.DagCBOR { return xerrors.New("invalid message prefix") } return nil }