cbjxtransport.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

    public boolean ping(EndpointAddress addr) {
        Messenger messenger = getMessenger(addr, null);

        boolean reachable = (null != messenger);

        if (messenger != null) {
            messenger.close();
        }

        return reachable;
    }

    /**
     * {@inheritDoc}
     */
    public void processIncomingMessage(Message message, EndpointAddress srcAddr, EndpointAddress dstAddr) {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("processIncomingMessage : Received message from: " + srcAddr);
        }

        // extract the Crypto info from the message
        MessageElement cryptoElement = message.getMessageElement(CBJX_MSG_NS, CBJX_MSG_INFO);

        if (cryptoElement == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("processIncomingMessage : No \'" + CBJX_MSG_INFO + "\' in the message");
            }
            return;
        }
        message.removeMessageElement(cryptoElement);

        // the cbjx message info
        CbJxMessageInfo cryptoInfo = null;

        try {
            cryptoInfo = new CbJxMessageInfo(cryptoElement.getStream(), cryptoElement.getMimeType());
        } catch (Throwable e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING
                        ,
                        "processIncomingMessage : Couldn\'t retrieve CbJxMessageInfo from \'" + CBJX_MSG_INFO + "\' element", e);
            }
            return;
        }

        Message submessage = checkCryptoInfo(message, cryptoElement, cryptoInfo);

        if (null == submessage) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("processIncomingMessage : discarding message from " + srcAddr);
            }
            return;
        }

        // give back the message to the endpoint
        try {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("processIncomingMessage: delivering " + submessage + " to: " + cryptoInfo.getDestinationAddress());
            }

            endpoint.processIncomingMessage(submessage, cryptoInfo.getSourceAddress(), cryptoInfo.getDestinationAddress());
        } catch (Throwable all) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "processIncomingMessage: endpoint failed to demux message", all);
            }
        }
    }

    /**
     * add the CryptoInfo into the message
     *
     * @param submessage  the message
     * @param destAddress the destination
     * @return Message the message with the CbJxMessageInfo added
     */
    public Message addCryptoInfo(Message submessage, EndpointAddress destAddress) throws IOException {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Building CBJX wrapper for " + submessage);
        }

        // Remove all existing CbJx Elements from source
        Iterator eachCbJxElement = submessage.getMessageElementsOfNamespace(CbJxTransport.CBJX_MSG_NS);

        while (eachCbJxElement.hasNext()) {
            MessageElement aMessageElement = (MessageElement) eachCbJxElement.next();

            eachCbJxElement.remove();
        }

        Message message = new Message();

        CbJxMessageInfo cryptoInfo = new CbJxMessageInfo();

        // set the source Id of the message
        cryptoInfo.setSourceID(localPeerID);
        cryptoInfo.setSourceAddress(localPeerAddr);
        cryptoInfo.setDestinationAddress(destAddress);

        // add the root cert into the message info
        PSECredential cred = (PSECredential) membership.getDefaultCredential();

        if (null == cred) {
            throw new IOException("No authentication available for message signing.");
        }

        Certificate cert = cred.getCertificate();

        cryptoInfo.setPeerCert(cert);

        // compute the signature of the message body
        TextDocument infoDoc = (TextDocument) cryptoInfo.getDocument(MimeMediaType.XMLUTF8);
        byte[] infoSignature = null;

        try {
            infoSignature = PSEUtils.computeSignature(CbJxDefs.signAlgoName, cred.getPrivateKey(), infoDoc.getStream());
        } catch (Throwable e) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "failed to sign " + submessage, e);
            }
            return null;
        }

        // add the cbjx:CryptoInfo into the message
        MessageElement infoSigElement = new ByteArrayMessageElement(CBJX_MSG_SIG, MimeMediaType.AOS, infoSignature, null);

        // add the cbjx:CryptoInfo into the message
        MessageElement cryptoInfoElement = new TextDocumentMessageElement(CBJX_MSG_INFO, infoDoc, infoSigElement);

        message.addMessageElement(CBJX_MSG_NS, cryptoInfoElement);

        // Compute the signature of the encapsulated message and append it to
        // the container.

        // serialize the container
        WireFormatMessage subserial = WireFormatMessageFactory.toWire(submessage, WireFormatMessageFactory.DEFAULT_WIRE_MIME, null);

        // calculate the signature
        byte[] bodySignature = null;

        try {
            bodySignature = PSEUtils.computeSignature(CbJxDefs.signAlgoName, cred.getPrivateKey(), subserial.getStream());
        } catch (Throwable e) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "failed to sign" + submessage, e);
            }
            return null;
        }

        subserial = null;

        // Make the signature into an element
        MessageElement bodySigElement = new ByteArrayMessageElement(CBJX_MSG_SIG, MimeMediaType.AOS, bodySignature, null);

        // Add the encapsulated body into the container message.
        message.addMessageElement(CBJX_MSG_NS
                ,
                new JxtaMessageMessageElement(CBJX_MSG_BODY, new MimeMediaType("application/x-jxta-msg"), submessage
                ,
                bodySigElement));

        return message;
    }

    public Message checkCryptoInfo(Message message, MessageElement cryptoElement, CbJxMessageInfo cryptoInfo) {

        // extract the body element  from the message
        JxtaMessageMessageElement bodyElement = (JxtaMessageMessageElement) message.getMessageElement(CBJX_MSG_NS, CBJX_MSG_BODY);

        if (null == bodyElement) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("No \'" + CBJX_MSG_BODY + "\' in " + message);
            }
            return null;
        }
        message.removeMessageElement(bodyElement);

        // extract the peer certificate
        Certificate peerCert = cryptoInfo.getPeerCert();

        // and from it the public key
        // the public key from the message
        RSAPublicKey publicKey = (RSAPublicKey) peerCert.getPublicKey();

        // check the cert validity
        try {
            peerCert.verify(publicKey);
        } catch (Exception e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Invalid peer cert", e);
            }
            return null;
        }

        // check the cbid
        try {
            net.jxta.impl.id.CBID.PeerID srcPeerID = (net.jxta.impl.id.CBID.PeerID) cryptoInfo.getSourceID();

            byte[] pub_der = peerCert.getPublicKey().getEncoded();
            net.jxta.impl.id.CBID.PeerID genID = (net.jxta.impl.id.CBID.PeerID) IDFactory.newPeerID(group.getPeerGroupID()
                    ,
                    pub_der);

            if (!srcPeerID.getUUID().equals(genID.getUUID())) {
                // the cbid is not valid. Discard the message
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.warning("CBID of " + message + " is not valid : " + srcPeerID + " != " + genID);
                }
                return null;
            }

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("CBID of the message is valid");
            }
        } catch (Throwable e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failed to verify cbid", e);
            }
            return null;
        }

        // verify the signature of the cryptinfo message
        try {
            boolean valid = PSEUtils.verifySignature(CbJxDefs.signAlgoName, peerCert, cryptoElement.getSignature().getBytes(false)
                    ,
                    cryptoElement.getStream());

            if (!valid) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.warning("Failed to verify the signature of cryptinfo for " + message);
                }
                return null;
            }
        } catch (Throwable e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "Failed to verify the signature of cryptinfo for " + message, e);
            }
            return null;
        }

        // then verify the signature
        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
            LOG.warning("verifying signature");
        }

        // verify the signature of the message
        try {
            boolean valid = PSEUtils.verifySignature(CbJxDefs.signAlgoName, peerCert, bodyElement.getSignature().getBytes(false)
                    ,
                    bodyElement.getStream());

            if (!valid) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.warning("failed to verify the signature of " + message);
                }
                return null;
            }
        } catch (Throwable e) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.log(Level.WARNING, "failed to verify the signature of " + message, e);
            }
            return null;
        }

        // the message is valid
        return bodyElement.getMessage();
    }

    /**
     * this class filters incoming messages.
     * it checks if messages are valid and if not discard them
     */
    public class CbJxInputFilter implements MessageFilterListener {
        public CbJxInputFilter() {
            super();
        }

        /**
         * {@inheritDoc}
         */
        public Message filterMessage(Message message, EndpointAddress srcAddr, EndpointAddress dstAddr) {

            if (dstAddr.getProtocolAddress().equals(getProtocolName())) {
                // extract the Crypto info from the message
                MessageElement cryptoElement = message.getMessageElement(CBJX_MSG_NS, CBJX_MSG_INFO);

                if (cryptoElement == null) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("No \'" + CBJX_MSG_INFO + "\' in the message");
                    }
                    return null;
                }
                message.removeMessageElement(cryptoElement);

                // the cbjx message info
                CbJxMessageInfo cryptoInfo = null;

                try {
                    cryptoInfo = new CbJxMessageInfo(cryptoElement.getStream(), cryptoElement.getMimeType());
                } catch (Throwable e) {
                    if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                        LOG.log(Level.WARNING, "Couldn\'t retrieve CbJxMessageInfo from \'" + CBJX_MSG_INFO + "\' element", e);
                    }
                    return null;
                }

                return checkCryptoInfo(message, cryptoElement, cryptoInfo);
            }

            return message;
        }
    }


    /**
     * this class filters all outgoing messages that are not sent with
     * messengers. (that is propagate messages). It adds CbJxInformation
     * into to messages.
     */
    public class CbJxOutputFilter implements MessageFilterListener {

        /**
         * Default constructor
         */
        public CbJxOutputFilter() {
            super();
        }

        /**
         * {@inheritDoc}
         */
        public Message filterMessage(Message message, EndpointAddress srcAddr, EndpointAddress dstAddr) {
            Message msg = message.clone();

            if (null == msg.getMessageElement(CBJX_MSG_NS, CBJX_MSG_INFO)) {
                try {
                    msg = addCryptoInfo(msg, dstAddr);
                } catch (IOException failed) {
                    return null;
                }
            }

            return msg;
        }
    }
}