HDDS-804. Block token: Add secret token manager. Contributed by Ajay Kumar.
This commit is contained in:
Ajay Kumar
Xiaoyu Yao
parent
0c8829a9a1
commit
6d522dc05d
|
|
@ -21,6 +21,7 @@
|
|||
|
||||
import com.google.common.base.Preconditions;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.ozone.OzoneConfigKeys;
|
||||
import org.bouncycastle.jce.provider.BouncyCastleProvider;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
|
@ -77,6 +78,7 @@ public class SecurityConfig {
|
|||
private final Duration certDuration;
|
||||
private final String x509SignatureAlgo;
|
||||
private final Boolean grpcBlockTokenEnabled;
|
||||
private final int getMaxKeyLength;
|
||||
private final String certificateDir;
|
||||
private final String certificateFileName;
|
||||
|
||||
|
|
@ -88,6 +90,9 @@ public class SecurityConfig {
|
|||
public SecurityConfig(Configuration configuration) {
|
||||
Preconditions.checkNotNull(configuration, "Configuration cannot be null");
|
||||
this.configuration = configuration;
|
||||
this.getMaxKeyLength = configuration.getInt(
|
||||
OzoneConfigKeys.OZONE_MAX_KEY_LEN,
|
||||
OzoneConfigKeys.OZONE_MAX_KEY_LEN_DEFAULT);
|
||||
this.size = this.configuration.getInt(HDDS_KEY_LEN, HDDS_DEFAULT_KEY_LEN);
|
||||
this.keyAlgo = this.configuration.get(HDDS_KEY_ALGORITHM,
|
||||
HDDS_DEFAULT_KEY_ALGORITHM);
|
||||
|
|
@ -289,4 +294,8 @@ private Provider initSecurityProvider(String providerName) {
|
|||
throw new SecurityException("Unknown security provider:" + provider);
|
||||
}
|
||||
}
|
||||
|
||||
public int getMaxKeyLength() {
|
||||
return this.getMaxKeyLength;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,191 @@
|
|||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with this
|
||||
* work for additional information regarding copyright ownership. The ASF
|
||||
* licenses this file to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
* <p>
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* <p>
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
* License for the specific language governing permissions and limitations under
|
||||
* the License.
|
||||
*/
|
||||
|
||||
package org.apache.hadoop.ozone.security;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.classification.InterfaceStability;
|
||||
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
|
||||
import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
|
||||
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.BlockTokenSecretProto.AccessModeProto;
|
||||
import org.apache.hadoop.io.Text;
|
||||
import org.apache.hadoop.security.UserGroupInformation;
|
||||
import org.apache.hadoop.security.token.Token;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.security.KeyPair;
|
||||
import java.util.EnumSet;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
/**
|
||||
* SecretManager for Ozone Master block tokens.
|
||||
*/
|
||||
@InterfaceAudience.Private
|
||||
@InterfaceStability.Unstable
|
||||
public class OzoneBlockTokenSecretManager extends
|
||||
OzoneSecretManager<OzoneBlockTokenIdentifier> {
|
||||
|
||||
private static final Logger LOG = LoggerFactory
|
||||
.getLogger(OzoneBlockTokenSecretManager.class);;
|
||||
// Will be set by grpc clients for individual datanodes.
|
||||
static final Text SERVICE = new Text("HDDS_SERVICE");
|
||||
private final String omCertSerialId;
|
||||
|
||||
/**
|
||||
* Create a secret manager.
|
||||
*
|
||||
* @param conf
|
||||
* @param blockTokenExpirytime token expiry time for expired tokens in
|
||||
* milliseconds
|
||||
*/
|
||||
public OzoneBlockTokenSecretManager(OzoneConfiguration conf,
|
||||
long blockTokenExpirytime, String omCertSerialId) {
|
||||
super(conf, blockTokenExpirytime, blockTokenExpirytime, SERVICE, LOG);
|
||||
this.omCertSerialId = omCertSerialId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public OzoneBlockTokenIdentifier createIdentifier() {
|
||||
throw new SecurityException("Ozone block token can't be created "
|
||||
+ "without owner and access mode information.");
|
||||
}
|
||||
|
||||
public OzoneBlockTokenIdentifier createIdentifier(String owner,
|
||||
String blockId, EnumSet<AccessModeProto> modes, long maxLength) {
|
||||
return new OzoneBlockTokenIdentifier(owner, blockId, modes,
|
||||
getTokenExpiryTime(), omCertSerialId, maxLength);
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate an block token for specified user, blockId.
|
||||
*
|
||||
* @param user
|
||||
* @param blockId
|
||||
* @param modes
|
||||
* @param maxLength
|
||||
* @return token
|
||||
*/
|
||||
public Token<OzoneBlockTokenIdentifier> generateToken(String user,
|
||||
String blockId, EnumSet<AccessModeProto> modes, long maxLength) {
|
||||
OzoneBlockTokenIdentifier tokenIdentifier = createIdentifier(user,
|
||||
blockId, modes, maxLength);
|
||||
if (LOG.isTraceEnabled()) {
|
||||
long expiryTime = tokenIdentifier.getExpiryDate();
|
||||
String tokenId = tokenIdentifier.toString();
|
||||
LOG.trace("Issued delegation token -> expiryTime:{},tokenId:{}",
|
||||
expiryTime, tokenId);
|
||||
}
|
||||
return new Token<>(tokenIdentifier.getBytes(),
|
||||
createPassword(tokenIdentifier), tokenIdentifier.getKind(), SERVICE);
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate an block token for current user.
|
||||
*
|
||||
* @param blockId
|
||||
* @param modes
|
||||
* @return token
|
||||
*/
|
||||
public Token<OzoneBlockTokenIdentifier> generateToken(String blockId,
|
||||
EnumSet<AccessModeProto> modes, long maxLength) throws IOException {
|
||||
UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
|
||||
String userID = (ugi == null ? null : ugi.getShortUserName());
|
||||
return generateToken(userID, blockId, modes, maxLength);
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] retrievePassword(OzoneBlockTokenIdentifier identifier)
|
||||
throws InvalidToken {
|
||||
validateToken(identifier);
|
||||
return createPassword(identifier);
|
||||
}
|
||||
|
||||
@Override
|
||||
public long renewToken(Token<OzoneBlockTokenIdentifier> token,
|
||||
String renewer) throws IOException {
|
||||
throw new UnsupportedOperationException("Renew token operation is not " +
|
||||
"supported for ozone block tokens.");
|
||||
}
|
||||
|
||||
@Override
|
||||
public OzoneBlockTokenIdentifier cancelToken(Token<OzoneBlockTokenIdentifier>
|
||||
token, String canceller) throws IOException {
|
||||
throw new UnsupportedOperationException("Cancel token operation is not " +
|
||||
"supported for ozone block tokens.");
|
||||
}
|
||||
|
||||
/**
|
||||
* Find the OzoneBlockTokenInfo for the given token id, and verify that if the
|
||||
* token is not expired.
|
||||
*/
|
||||
public boolean validateToken(OzoneBlockTokenIdentifier identifier)
|
||||
throws InvalidToken {
|
||||
long now = Time.now();
|
||||
if (identifier.getExpiryDate() < now) {
|
||||
throw new InvalidToken("token " + formatTokenId(identifier) + " is " +
|
||||
"expired, current time: " + Time.formatTime(now) +
|
||||
" expiry time: " + identifier.getExpiryDate());
|
||||
}
|
||||
|
||||
if (!verifySignature(identifier, createPassword(identifier))) {
|
||||
throw new InvalidToken("Tampared/Inavalid token.");
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Should be called before this object is used.
|
||||
*/
|
||||
@Override
|
||||
public synchronized void start(KeyPair keyPair) throws IOException {
|
||||
super.start(keyPair);
|
||||
removeExpiredKeys();
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns expiry time by adding configured expiry time with current time.
|
||||
*
|
||||
* @return Expiry time.
|
||||
*/
|
||||
private long getTokenExpiryTime() {
|
||||
return Time.now() + getTokenRenewInterval();
|
||||
}
|
||||
|
||||
/**
|
||||
* Should be called before this object is used.
|
||||
*/
|
||||
@Override
|
||||
public synchronized void stop() throws IOException {
|
||||
super.stop();
|
||||
}
|
||||
|
||||
private synchronized void removeExpiredKeys() {
|
||||
// TODO: handle roll private key/certificate
|
||||
long now = Time.now();
|
||||
for (Iterator<Map.Entry<Integer, OzoneSecretKey>> it = allKeys.entrySet()
|
||||
.iterator(); it.hasNext();) {
|
||||
Map.Entry<Integer, OzoneSecretKey> e = it.next();
|
||||
OzoneSecretKey key = e.getValue();
|
||||
if (key.getExpiryDate() < now) {
|
||||
it.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,455 @@
|
|||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||
* contributor license agreements. See the NOTICE file distributed with this
|
||||
* work for additional information regarding copyright ownership. The ASF
|
||||
* licenses this file to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
* <p>
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* <p>
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
* License for the specific language governing permissions and limitations under
|
||||
* the License.
|
||||
*/
|
||||
|
||||
package org.apache.hadoop.ozone.security;
|
||||
|
||||
import org.apache.hadoop.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.classification.InterfaceStability;
|
||||
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
|
||||
import org.apache.hadoop.io.Text;
|
||||
import org.apache.hadoop.ozone.security.OzoneSecretStore.OzoneManagerSecretState;
|
||||
import org.apache.hadoop.ozone.security.OzoneTokenIdentifier.TokenInfo;
|
||||
import org.apache.hadoop.security.AccessControlException;
|
||||
import org.apache.hadoop.security.HadoopKerberosName;
|
||||
import org.apache.hadoop.security.token.Token;
|
||||
import org.apache.hadoop.util.Daemon;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.DataInputStream;
|
||||
import java.io.IOException;
|
||||
import java.security.KeyPair;
|
||||
import java.security.PrivateKey;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* SecretManager for Ozone Master. Responsible for signing identifiers with
|
||||
* private key,
|
||||
*/
|
||||
@InterfaceAudience.Private
|
||||
@InterfaceStability.Unstable
|
||||
public class OzoneDelegationTokenSecretManager<T extends OzoneTokenIdentifier>
|
||||
extends OzoneSecretManager<T> {
|
||||
|
||||
private static final Logger LOG = LoggerFactory
|
||||
.getLogger(OzoneDelegationTokenSecretManager.class);
|
||||
private final Map<T, TokenInfo> currentTokens;
|
||||
private final OzoneSecretStore store;
|
||||
private Thread tokenRemoverThread;
|
||||
private final long tokenRemoverScanInterval;
|
||||
/**
|
||||
* If the delegation token update thread holds this lock, it will not get
|
||||
* interrupted.
|
||||
*/
|
||||
private Object noInterruptsLock = new Object();
|
||||
|
||||
/**
|
||||
* Create a secret manager.
|
||||
*
|
||||
* @param conf configuration.
|
||||
* @param tokenMaxLifetime the maximum lifetime of the delegation tokens in
|
||||
* milliseconds
|
||||
* @param tokenRenewInterval how often the tokens must be renewed in
|
||||
* milliseconds
|
||||
* @param dtRemoverScanInterval how often the tokens are scanned for expired
|
||||
* tokens in milliseconds
|
||||
*/
|
||||
public OzoneDelegationTokenSecretManager(OzoneConfiguration conf,
|
||||
long tokenMaxLifetime, long tokenRenewInterval,
|
||||
long dtRemoverScanInterval, Text service) throws IOException {
|
||||
super(conf, tokenMaxLifetime, tokenRenewInterval, service, LOG);
|
||||
currentTokens = new ConcurrentHashMap();
|
||||
this.tokenRemoverScanInterval = dtRemoverScanInterval;
|
||||
this.store = new OzoneSecretStore(conf);
|
||||
loadTokenSecretState(store.loadState());
|
||||
}
|
||||
|
||||
@Override
|
||||
public T createIdentifier() {
|
||||
return (T) T.newInstance();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create new Identifier with given,owner,renwer and realUser.
|
||||
*
|
||||
* @return T
|
||||
*/
|
||||
public T createIdentifier(Text owner, Text renewer, Text realUser) {
|
||||
return (T) T.newInstance(owner, renewer, realUser);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns {@link Token} for given identifier.
|
||||
*
|
||||
* @param owner
|
||||
* @param renewer
|
||||
* @param realUser
|
||||
* @return Token
|
||||
* @throws IOException to allow future exceptions to be added without breaking
|
||||
* compatibility
|
||||
*/
|
||||
public Token<T> createToken(Text owner, Text renewer, Text realUser)
|
||||
throws IOException {
|
||||
T identifier = createIdentifier(owner, renewer, realUser);
|
||||
updateIdentifierDetails(identifier);
|
||||
|
||||
byte[] password = createPassword(identifier.getBytes(),
|
||||
getCurrentKey().getPrivateKey());
|
||||
addToTokenStore(identifier, password);
|
||||
Token<T> token = new Token<>(identifier.getBytes(), password,
|
||||
identifier.getKind(), getService());
|
||||
if (LOG.isTraceEnabled()) {
|
||||
long expiryTime = identifier.getIssueDate() + getTokenRenewInterval();
|
||||
String tokenId = identifier.toStringStable();
|
||||
LOG.trace("Issued delegation token -> expiryTime:{},tokenId:{}",
|
||||
expiryTime, tokenId);
|
||||
}
|
||||
return token;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stores given identifier in token store.
|
||||
*
|
||||
* @param identifier
|
||||
* @param password
|
||||
* @throws IOException
|
||||
*/
|
||||
private void addToTokenStore(T identifier, byte[] password)
|
||||
throws IOException {
|
||||
TokenInfo tokenInfo = new TokenInfo(identifier.getIssueDate()
|
||||
+ getTokenRenewInterval(), password, identifier.getTrackingId());
|
||||
currentTokens.put(identifier, tokenInfo);
|
||||
store.storeToken(identifier, tokenInfo.getRenewDate());
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates issue date, master key id and sequence number for identifier.
|
||||
*
|
||||
* @param identifier the identifier to validate
|
||||
*/
|
||||
private void updateIdentifierDetails(T identifier) {
|
||||
int sequenceNum;
|
||||
long now = Time.monotonicNow();
|
||||
sequenceNum = incrementDelegationTokenSeqNum();
|
||||
identifier.setIssueDate(now);
|
||||
identifier.setMasterKeyId(getCurrentKey().getKeyId());
|
||||
identifier.setSequenceNumber(sequenceNum);
|
||||
identifier.setMaxDate(Time.monotonicNow() + getTokenMaxLifetime());
|
||||
}
|
||||
|
||||
/**
|
||||
* Renew a delegation token.
|
||||
*
|
||||
* @param token the token to renew
|
||||
* @param renewer the full principal name of the user doing the renewal
|
||||
* @return the new expiration time
|
||||
* @throws InvalidToken if the token is invalid
|
||||
* @throws AccessControlException if the user can't renew token
|
||||
*/
|
||||
@Override
|
||||
public synchronized long renewToken(Token<T> token, String renewer)
|
||||
throws IOException {
|
||||
ByteArrayInputStream buf = new ByteArrayInputStream(token.getIdentifier());
|
||||
DataInputStream in = new DataInputStream(buf);
|
||||
T id = (T) T.readProtoBuf(in);
|
||||
if(LOG.isDebugEnabled()) {
|
||||
LOG.debug("Token renewal for identifier: {}, total currentTokens: {}",
|
||||
formatTokenId(id), currentTokens.size());
|
||||
}
|
||||
|
||||
long now = Time.monotonicNow();
|
||||
if (id.getMaxDate() < now) {
|
||||
throw new InvalidToken(renewer + " tried to renew an expired token "
|
||||
+ formatTokenId(id) + " max expiration date: "
|
||||
+ Time.formatTime(id.getMaxDate())
|
||||
+ " currentTime: " + Time.formatTime(now));
|
||||
}
|
||||
validateToken(id);
|
||||
if ((id.getRenewer() == null) || (id.getRenewer().toString().isEmpty())) {
|
||||
throw new AccessControlException(renewer +
|
||||
" tried to renew a token " + formatTokenId(id)
|
||||
+ " without a renewer");
|
||||
}
|
||||
if (!id.getRenewer().toString().equals(renewer)) {
|
||||
throw new AccessControlException(renewer
|
||||
+ " tries to renew a token " + formatTokenId(id)
|
||||
+ " with non-matching renewer " + id.getRenewer());
|
||||
}
|
||||
OzoneSecretKey key = allKeys.get(id.getMasterKeyId());
|
||||
if (key == null) {
|
||||
throw new InvalidToken("Unable to find master key for keyId="
|
||||
+ id.getMasterKeyId()
|
||||
+ " from cache. Failed to renew an unexpired token "
|
||||
+ formatTokenId(id) + " with sequenceNumber="
|
||||
+ id.getSequenceNumber());
|
||||
}
|
||||
byte[] password = createPassword(token.getIdentifier(),
|
||||
key.getPrivateKey());
|
||||
|
||||
long renewTime = Math.min(id.getMaxDate(), now + getTokenRenewInterval());
|
||||
try {
|
||||
addToTokenStore(id, password);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Unable to update token " + id.getSequenceNumber(), e);
|
||||
}
|
||||
return renewTime;
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel a token by removing it from store and cache.
|
||||
*
|
||||
* @return Identifier of the canceled token
|
||||
* @throws InvalidToken for invalid token
|
||||
* @throws AccessControlException if the user isn't allowed to cancel
|
||||
*/
|
||||
public T cancelToken(Token<T> token, String canceller) throws IOException {
|
||||
T id = (T) T.readProtoBuf(token.getIdentifier());
|
||||
LOG.debug("Token cancellation requested for identifier: {}",
|
||||
formatTokenId(id));
|
||||
|
||||
if (id.getUser() == null) {
|
||||
throw new InvalidToken("Token with no owner " + formatTokenId(id));
|
||||
}
|
||||
String owner = id.getUser().getUserName();
|
||||
Text renewer = id.getRenewer();
|
||||
HadoopKerberosName cancelerKrbName = new HadoopKerberosName(canceller);
|
||||
String cancelerShortName = cancelerKrbName.getShortName();
|
||||
if (!canceller.equals(owner)
|
||||
&& (renewer == null || renewer.toString().isEmpty()
|
||||
|| !cancelerShortName
|
||||
.equals(renewer.toString()))) {
|
||||
throw new AccessControlException(canceller
|
||||
+ " is not authorized to cancel the token " + formatTokenId(id));
|
||||
}
|
||||
try {
|
||||
store.removeToken(id);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Unable to remove token " + id.getSequenceNumber(), e);
|
||||
}
|
||||
TokenInfo info = currentTokens.remove(id);
|
||||
if (info == null) {
|
||||
throw new InvalidToken("Token not found " + formatTokenId(id));
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] retrievePassword(T identifier) throws InvalidToken {
|
||||
return validateToken(identifier).getPassword();
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if TokenInfo for the given identifier exists in database and if the
|
||||
* token is expired.
|
||||
*/
|
||||
public TokenInfo validateToken(T identifier) throws InvalidToken {
|
||||
TokenInfo info = currentTokens.get(identifier);
|
||||
if (info == null) {
|
||||
throw new InvalidToken("token " + formatTokenId(identifier)
|
||||
+ " can't be found in cache");
|
||||
}
|
||||
long now = Time.monotonicNow();
|
||||
if (info.getRenewDate() < now) {
|
||||
throw new InvalidToken("token " + formatTokenId(identifier) + " is " +
|
||||
"expired, current time: " + Time.formatTime(now) +
|
||||
" expected renewal time: " + Time.formatTime(info.getRenewDate()));
|
||||
}
|
||||
if (!verifySignature(identifier, info.getPassword())) {
|
||||
throw new InvalidToken("Tampared/Inavalid token.");
|
||||
}
|
||||
return info;
|
||||
}
|
||||
|
||||
// TODO: handle roll private key/certificate
|
||||
private synchronized void removeExpiredKeys() {
|
||||
long now = Time.monotonicNow();
|
||||
for (Iterator<Map.Entry<Integer, OzoneSecretKey>> it = allKeys.entrySet()
|
||||
.iterator(); it.hasNext();) {
|
||||
Map.Entry<Integer, OzoneSecretKey> e = it.next();
|
||||
OzoneSecretKey key = e.getValue();
|
||||
if (key.getExpiryDate() < now && key.getExpiryDate() != -1) {
|
||||
if (!key.equals(getCurrentKey())) {
|
||||
it.remove();
|
||||
try {
|
||||
store.removeTokenMasterKey(key);
|
||||
} catch (IOException ex) {
|
||||
LOG.error("Unable to remove master key " + key.getKeyId(), ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void loadTokenSecretState(OzoneManagerSecretState<T> state)
|
||||
throws IOException {
|
||||
LOG.info("Loading token state into token manager.");
|
||||
for (OzoneSecretKey key : state.ozoneManagerSecretState()) {
|
||||
allKeys.putIfAbsent(key.getKeyId(), key);
|
||||
incrementCurrentKeyId();
|
||||
}
|
||||
for (Map.Entry<T, Long> entry : state.getTokenState().entrySet()) {
|
||||
addPersistedDelegationToken(entry.getKey(), entry.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
private void addPersistedDelegationToken(
|
||||
T identifier, long renewDate)
|
||||
throws IOException {
|
||||
if (isRunning()) {
|
||||
// a safety check
|
||||
throw new IOException(
|
||||
"Can't add persisted delegation token to a running SecretManager.");
|
||||
}
|
||||
int keyId = identifier.getMasterKeyId();
|
||||
OzoneSecretKey dKey = allKeys.get(keyId);
|
||||
if (dKey == null) {
|
||||
LOG.warn("No KEY found for persisted identifier "
|
||||
+ formatTokenId(identifier));
|
||||
return;
|
||||
}
|
||||
|
||||
PrivateKey privateKey = dKey.getPrivateKey();
|
||||
byte[] password = createPassword(identifier.getBytes(), privateKey);
|
||||
if (identifier.getSequenceNumber() > getDelegationTokenSeqNum()) {
|
||||
setDelegationTokenSeqNum(identifier.getSequenceNumber());
|
||||
}
|
||||
if (currentTokens.get(identifier) == null) {
|
||||
currentTokens.put(identifier, new TokenInfo(renewDate,
|
||||
password, identifier.getTrackingId()));
|
||||
} else {
|
||||
throw new IOException("Same delegation token being added twice: "
|
||||
+ formatTokenId(identifier));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Should be called before this object is used.
|
||||
*/
|
||||
@Override
|
||||
public synchronized void start(KeyPair keyPair) throws IOException {
|
||||
super.start(keyPair);
|
||||
storeKey(getCurrentKey());
|
||||
removeExpiredKeys();
|
||||
tokenRemoverThread = new Daemon(new ExpiredTokenRemover());
|
||||
tokenRemoverThread.start();
|
||||
}
|
||||
|
||||
private void storeKey(OzoneSecretKey key) throws IOException {
|
||||
store.storeTokenMasterKey(key);
|
||||
if (!allKeys.containsKey(key.getKeyId())) {
|
||||
allKeys.put(key.getKeyId(), key);
|
||||
}
|
||||
}
|
||||
|
||||
public void stopThreads() {
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("Stopping expired delegation token remover thread");
|
||||
}
|
||||
setIsRunning(false);
|
||||
|
||||
if (tokenRemoverThread != null) {
|
||||
synchronized (noInterruptsLock) {
|
||||
tokenRemoverThread.interrupt();
|
||||
}
|
||||
try {
|
||||
tokenRemoverThread.join();
|
||||
} catch (InterruptedException e) {
|
||||
throw new RuntimeException(
|
||||
"Unable to join on token removal thread", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stops the OzoneDelegationTokenSecretManager.
|
||||
*
|
||||
* @throws IOException
|
||||
*/
|
||||
@Override
|
||||
public void stop() throws IOException {
|
||||
super.stop();
|
||||
stopThreads();
|
||||
if (this.store != null) {
|
||||
this.store.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove expired delegation tokens from cache and persisted store.
|
||||
*/
|
||||
private void removeExpiredToken() {
|
||||
long now = Time.monotonicNow();
|
||||
synchronized (this) {
|
||||
Iterator<Map.Entry<T,
|
||||
TokenInfo>> i = currentTokens.entrySet().iterator();
|
||||
while (i.hasNext()) {
|
||||
Map.Entry<T,
|
||||
TokenInfo> entry = i.next();
|
||||
long renewDate = entry.getValue().getRenewDate();
|
||||
if (renewDate < now) {
|
||||
i.remove();
|
||||
try {
|
||||
store.removeToken(entry.getKey());
|
||||
} catch (IOException e) {
|
||||
if(LOG.isDebugEnabled()) {
|
||||
LOG.debug("Failed to remove expired token {}", entry.getValue());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private class ExpiredTokenRemover extends Thread {
|
||||
private long lastTokenCacheCleanup;
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
LOG.info("Starting expired delegation token remover thread, "
|
||||
+ "tokenRemoverScanInterval=" + getTokenRemoverScanInterval()
|
||||
/ (60 * 1000) + " min(s)");
|
||||
try {
|
||||
while (isRunning()) {
|
||||
long now = Time.monotonicNow();
|
||||
if (lastTokenCacheCleanup + getTokenRemoverScanInterval()
|
||||
< now) {
|
||||
removeExpiredToken();
|
||||
lastTokenCacheCleanup = now;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(Math.min(5000,
|
||||
getTokenRemoverScanInterval())); // 5 seconds
|
||||
} catch (InterruptedException ie) {
|
||||
LOG.error("ExpiredTokenRemover received " + ie);
|
||||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
LOG.error("ExpiredTokenRemover thread received unexpected exception",
|
||||
t);
|
||||
Runtime.getRuntime().exit(-1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public long getTokenRemoverScanInterval() {
|
||||
return tokenRemoverScanInterval;
|
||||
}
|
||||
}
|
||||
|
|
@ -18,8 +18,17 @@
|
|||
package org.apache.hadoop.ozone.security;
|
||||
|
||||
import com.google.common.base.Preconditions;
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.DataInputStream;
|
||||
import org.apache.hadoop.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.classification.InterfaceStability;
|
||||
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
|
||||
import org.apache.hadoop.hdds.security.x509.SecurityConfig;
|
||||
import org.apache.hadoop.io.Text;
|
||||
import org.apache.hadoop.security.AccessControlException;
|
||||
import org.apache.hadoop.security.token.SecretManager;
|
||||
import org.apache.hadoop.security.token.Token;
|
||||
import org.apache.hadoop.security.token.TokenIdentifier;
|
||||
import org.slf4j.Logger;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.security.InvalidKeyException;
|
||||
import java.security.KeyPair;
|
||||
|
|
@ -27,25 +36,9 @@
|
|||
import java.security.PrivateKey;
|
||||
import java.security.Signature;
|
||||
import java.security.SignatureException;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import org.apache.hadoop.classification.InterfaceAudience;
|
||||
import org.apache.hadoop.classification.InterfaceStability;
|
||||
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
|
||||
import org.apache.hadoop.io.Text;
|
||||
import org.apache.hadoop.ozone.OzoneConfigKeys;
|
||||
import org.apache.hadoop.ozone.security.OzoneSecretStore.OzoneManagerSecretState;
|
||||
import org.apache.hadoop.ozone.security.OzoneTokenIdentifier.TokenInfo;
|
||||
import org.apache.hadoop.security.AccessControlException;
|
||||
import org.apache.hadoop.security.HadoopKerberosName;
|
||||
import org.apache.hadoop.security.token.SecretManager;
|
||||
import org.apache.hadoop.security.token.Token;
|
||||
import org.apache.hadoop.util.Daemon;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* SecretManager for Ozone Master. Responsible for signing identifiers with
|
||||
|
|
@ -53,33 +46,23 @@
|
|||
*/
|
||||
@InterfaceAudience.Private
|
||||
@InterfaceStability.Unstable
|
||||
public class OzoneSecretManager<T extends OzoneTokenIdentifier>
|
||||
public abstract class OzoneSecretManager<T extends TokenIdentifier>
|
||||
extends SecretManager<T> {
|
||||
|
||||
private static final Logger LOG = LoggerFactory
|
||||
.getLogger(OzoneSecretManager.class);
|
||||
private final Logger logger;
|
||||
/**
|
||||
* The name of the Private/Public Key based hashing algorithm.
|
||||
*/
|
||||
private static final String DEFAULT_SIGNATURE_ALGORITHM = "SHA256withRSA";
|
||||
private final SecurityConfig securityConfig;
|
||||
private final long tokenMaxLifetime;
|
||||
private final long tokenRenewInterval;
|
||||
private final long tokenRemoverScanInterval;
|
||||
private final Text service;
|
||||
private final Map<Integer, OzoneSecretKey> allKeys;
|
||||
private final Map<T, TokenInfo> currentTokens;
|
||||
private final OzoneSecretStore store;
|
||||
private Thread tokenRemoverThread;
|
||||
private volatile boolean running;
|
||||
private AtomicInteger tokenSequenceNumber;
|
||||
private OzoneSecretKey currentKey;
|
||||
private AtomicInteger currentKeyId;
|
||||
/**
|
||||
* If the delegation token update thread holds this lock, it will not get
|
||||
* interrupted.
|
||||
*/
|
||||
private Object noInterruptsLock = new Object();
|
||||
private int maxKeyLength;
|
||||
private AtomicInteger currentKeyId;
|
||||
private AtomicInteger tokenSequenceNumber;
|
||||
protected final Map<Integer, OzoneSecretKey> allKeys;
|
||||
|
||||
/**
|
||||
* Create a secret manager.
|
||||
|
|
@ -89,100 +72,21 @@ public class OzoneSecretManager<T extends OzoneTokenIdentifier>
|
|||
* milliseconds
|
||||
* @param tokenRenewInterval how often the tokens must be renewed in
|
||||
* milliseconds
|
||||
* @param dtRemoverScanInterval how often the tokens are scanned for expired
|
||||
* tokens in milliseconds
|
||||
* @param service name of service
|
||||
*/
|
||||
public OzoneSecretManager(OzoneConfiguration conf, long tokenMaxLifetime,
|
||||
long tokenRenewInterval, long dtRemoverScanInterval, Text service)
|
||||
throws IOException {
|
||||
long tokenRenewInterval, Text service, Logger logger) {
|
||||
this.securityConfig = new SecurityConfig(conf);
|
||||
this.tokenMaxLifetime = tokenMaxLifetime;
|
||||
this.tokenRenewInterval = tokenRenewInterval;
|
||||
this.tokenRemoverScanInterval = dtRemoverScanInterval;
|
||||
|
||||
currentTokens = new ConcurrentHashMap();
|
||||
allKeys = new ConcurrentHashMap<>();
|
||||
currentKeyId = new AtomicInteger();
|
||||
tokenSequenceNumber = new AtomicInteger();
|
||||
this.store = new OzoneSecretStore(conf);
|
||||
loadTokenSecretState(store.loadState());
|
||||
allKeys = new ConcurrentHashMap<>();
|
||||
this.service = service;
|
||||
this.maxKeyLength = conf.getInt(OzoneConfigKeys.OZONE_MAX_KEY_LEN,
|
||||
OzoneConfigKeys.OZONE_MAX_KEY_LEN_DEFAULT);
|
||||
this.maxKeyLength = securityConfig.getMaxKeyLength();
|
||||
this.logger = logger;
|
||||
}
|
||||
|
||||
@Override
|
||||
public T createIdentifier() {
|
||||
return (T) T.newInstance();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create new Identifier with given,owner,renwer and realUser.
|
||||
*
|
||||
* @return T
|
||||
*/
|
||||
public T createIdentifier(Text owner, Text renewer, Text realUser) {
|
||||
return (T) T.newInstance(owner, renewer, realUser);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns {@link Token} for given identifier.
|
||||
*
|
||||
* @param owner
|
||||
* @param renewer
|
||||
* @param realUser
|
||||
* @return Token
|
||||
* @throws IOException to allow future exceptions to be added without breaking
|
||||
* compatibility
|
||||
*/
|
||||
public Token<T> createToken(Text owner, Text renewer, Text realUser)
|
||||
throws IOException {
|
||||
T identifier = createIdentifier(owner, renewer, realUser);
|
||||
updateIdentifierDetails(identifier);
|
||||
|
||||
byte[] password = createPassword(identifier.getBytes(),
|
||||
currentKey.getPrivateKey());
|
||||
addToTokenStore(identifier, password);
|
||||
Token<T> token = new Token<>(identifier.getBytes(), password,
|
||||
identifier.getKind(), service);
|
||||
if (LOG.isTraceEnabled()) {
|
||||
long expiryTime = identifier.getIssueDate() + tokenRenewInterval;
|
||||
String tokenId = identifier.toStringStable();
|
||||
LOG.trace("Issued delegation token -> expiryTime:{},tokenId:{}",
|
||||
expiryTime, tokenId);
|
||||
}
|
||||
|
||||
return token;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stores given identifier in token store.
|
||||
*
|
||||
* @param identifier
|
||||
* @param password
|
||||
* @throws IOException
|
||||
*/
|
||||
private void addToTokenStore(T identifier, byte[] password)
|
||||
throws IOException {
|
||||
TokenInfo tokenInfo = new TokenInfo(identifier.getIssueDate()
|
||||
+ tokenRenewInterval, password, identifier.getTrackingId());
|
||||
currentTokens.put(identifier, tokenInfo);
|
||||
store.storeToken(identifier, tokenInfo.getRenewDate());
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates issue date, master key id and sequence number for identifier.
|
||||
*
|
||||
* @param identifier the identifier to validate
|
||||
*/
|
||||
private void updateIdentifierDetails(T identifier) {
|
||||
int sequenceNum;
|
||||
long now = Time.monotonicNow();
|
||||
sequenceNum = incrementDelegationTokenSeqNum();
|
||||
identifier.setIssueDate(now);
|
||||
identifier.setMasterKeyId(currentKey.getKeyId());
|
||||
identifier.setSequenceNumber(sequenceNum);
|
||||
identifier.setMaxDate(Time.monotonicNow() + tokenMaxLifetime);
|
||||
}
|
||||
|
||||
/**
|
||||
* Compute HMAC of the identifier using the private key and return the output
|
||||
|
|
@ -196,7 +100,7 @@ public byte[] createPassword(byte[] identifier, PrivateKey privateKey)
|
|||
throws OzoneSecurityException {
|
||||
try {
|
||||
Signature rsaSignature = Signature.getInstance(
|
||||
DEFAULT_SIGNATURE_ALGORITHM);
|
||||
getDefaultSignatureAlgorithm());
|
||||
rsaSignature.initSign(privateKey);
|
||||
rsaSignature.update(identifier);
|
||||
return rsaSignature.sign();
|
||||
|
|
@ -210,22 +114,31 @@ public byte[] createPassword(byte[] identifier, PrivateKey privateKey)
|
|||
|
||||
@Override
|
||||
public byte[] createPassword(T identifier) {
|
||||
LOG.debug("Creating password for identifier: {}, currentKey: {}",
|
||||
logger.debug("Creating password for identifier: {}, currentKey: {}",
|
||||
formatTokenId(identifier), currentKey.getKeyId());
|
||||
byte[] password = null;
|
||||
try {
|
||||
password = createPassword(identifier.getBytes(),
|
||||
currentKey.getPrivateKey());
|
||||
} catch (IOException ioe) {
|
||||
LOG.error("Could not store token {}!!", formatTokenId(identifier),
|
||||
logger.error("Could not store token {}!!", formatTokenId(identifier),
|
||||
ioe);
|
||||
}
|
||||
return password;
|
||||
}
|
||||
|
||||
/**
|
||||
* Default implementation for Ozone. Verifies if hash in token is legit.
|
||||
* */
|
||||
@Override
|
||||
public byte[] retrievePassword(T identifier) throws InvalidToken {
|
||||
return checkToken(identifier).getPassword();
|
||||
byte[] password = createPassword(identifier);
|
||||
// TODO: Revisit this when key/certificate rotation is implemented.
|
||||
// i.e Try all valid keys instead of current key only.
|
||||
if (!verifySignature(identifier, password)) {
|
||||
throw new InvalidToken("Tampared/Inavalid token.");
|
||||
}
|
||||
return password;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -237,52 +150,8 @@ public byte[] retrievePassword(T identifier) throws InvalidToken {
|
|||
* @throws InvalidToken if the token is invalid
|
||||
* @throws AccessControlException if the user can't renew token
|
||||
*/
|
||||
public synchronized long renewToken(Token<T> token, String renewer)
|
||||
throws IOException {
|
||||
ByteArrayInputStream buf = new ByteArrayInputStream(token.getIdentifier());
|
||||
DataInputStream in = new DataInputStream(buf);
|
||||
T id = (T) T.readProtoBuf(in);
|
||||
LOG.debug("Token renewal for identifier: {}, total currentTokens: {}",
|
||||
formatTokenId(id), currentTokens.size());
|
||||
|
||||
long now = Time.monotonicNow();
|
||||
if (id.getMaxDate() < now) {
|
||||
throw new InvalidToken(renewer + " tried to renew an expired token "
|
||||
+ formatTokenId(id) + " max expiration date: "
|
||||
+ Time.formatTime(id.getMaxDate())
|
||||
+ " currentTime: " + Time.formatTime(now));
|
||||
}
|
||||
checkToken(id);
|
||||
if ((id.getRenewer() == null) || (id.getRenewer().toString().isEmpty())) {
|
||||
throw new AccessControlException(renewer +
|
||||
" tried to renew a token " + formatTokenId(id)
|
||||
+ " without a renewer");
|
||||
}
|
||||
if (!id.getRenewer().toString().equals(renewer)) {
|
||||
throw new AccessControlException(renewer
|
||||
+ " tries to renew a token " + formatTokenId(id)
|
||||
+ " with non-matching renewer " + id.getRenewer());
|
||||
}
|
||||
OzoneSecretKey key = allKeys.get(id.getMasterKeyId());
|
||||
if (key == null) {
|
||||
throw new InvalidToken("Unable to find master key for keyId="
|
||||
+ id.getMasterKeyId()
|
||||
+ " from cache. Failed to renew an unexpired token "
|
||||
+ formatTokenId(id) + " with sequenceNumber="
|
||||
+ id.getSequenceNumber());
|
||||
}
|
||||
byte[] password = createPassword(token.getIdentifier(),
|
||||
key.getPrivateKey());
|
||||
|
||||
long renewTime = Math.min(id.getMaxDate(), now + tokenRenewInterval);
|
||||
try {
|
||||
addToTokenStore(id, password);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Unable to update token " + id.getSequenceNumber(), e);
|
||||
}
|
||||
return renewTime;
|
||||
}
|
||||
|
||||
public abstract long renewToken(Token<T> token, String renewer)
|
||||
throws IOException;
|
||||
/**
|
||||
* Cancel a token by removing it from store and cache.
|
||||
*
|
||||
|
|
@ -290,44 +159,8 @@ public synchronized long renewToken(Token<T> token, String renewer)
|
|||
* @throws InvalidToken for invalid token
|
||||
* @throws AccessControlException if the user isn't allowed to cancel
|
||||
*/
|
||||
public T cancelToken(Token<T> token, String canceller) throws IOException {
|
||||
T id = (T) T.readProtoBuf(token.getIdentifier());
|
||||
LOG.debug("Token cancellation requested for identifier: {}",
|
||||
formatTokenId(id));
|
||||
|
||||
if (id.getUser() == null) {
|
||||
throw new InvalidToken("Token with no owner " + formatTokenId(id));
|
||||
}
|
||||
String owner = id.getUser().getUserName();
|
||||
Text renewer = id.getRenewer();
|
||||
HadoopKerberosName cancelerKrbName = new HadoopKerberosName(canceller);
|
||||
String cancelerShortName = cancelerKrbName.getShortName();
|
||||
if (!canceller.equals(owner)
|
||||
&& (renewer == null || renewer.toString().isEmpty()
|
||||
|| !cancelerShortName
|
||||
.equals(renewer.toString()))) {
|
||||
throw new AccessControlException(canceller
|
||||
+ " is not authorized to cancel the token " + formatTokenId(id));
|
||||
}
|
||||
try {
|
||||
store.removeToken(id);
|
||||
} catch (IOException e) {
|
||||
LOG.error("Unable to remove token " + id.getSequenceNumber(), e);
|
||||
}
|
||||
TokenInfo info = currentTokens.remove(id);
|
||||
if (info == null) {
|
||||
throw new InvalidToken("Token not found " + formatTokenId(id));
|
||||
}
|
||||
return id;
|
||||
}
|
||||
|
||||
public int getCurrentKeyId() {
|
||||
return currentKeyId.get();
|
||||
}
|
||||
|
||||
public void setCurrentKeyId(int keyId) {
|
||||
currentKeyId.set(keyId);
|
||||
}
|
||||
public abstract T cancelToken(Token<T> token, String canceller)
|
||||
throws IOException;
|
||||
|
||||
public int incrementCurrentKeyId() {
|
||||
return currentKeyId.incrementAndGet();
|
||||
|
|
@ -346,14 +179,31 @@ public int incrementDelegationTokenSeqNum() {
|
|||
}
|
||||
|
||||
/**
|
||||
* Validates if given token is valid.
|
||||
* Update the current master key. This is called once by start method before
|
||||
* tokenRemoverThread is created,
|
||||
*/
|
||||
private OzoneSecretKey updateCurrentKey(KeyPair keyPair) throws IOException {
|
||||
logger.info("Updating the current master key for generating tokens");
|
||||
|
||||
// TODO: fix me based on the certificate expire time to set the key
|
||||
// expire time.
|
||||
int newCurrentId = incrementCurrentKeyId();
|
||||
OzoneSecretKey newKey = new OzoneSecretKey(newCurrentId, -1,
|
||||
keyPair, maxKeyLength);
|
||||
currentKey = newKey;
|
||||
return currentKey;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates if given hash is valid.
|
||||
*
|
||||
* @param identifier
|
||||
* @param password
|
||||
*/
|
||||
private boolean validateToken(T identifier, byte[] password) {
|
||||
public boolean verifySignature(T identifier, byte[] password) {
|
||||
try {
|
||||
Signature rsaSignature = Signature.getInstance("SHA256withRSA");
|
||||
Signature rsaSignature =
|
||||
Signature.getInstance(getDefaultSignatureAlgorithm());
|
||||
rsaSignature.initVerify(currentKey.getPublicKey());
|
||||
rsaSignature.update(identifier.getBytes());
|
||||
return rsaSignature.verify(password);
|
||||
|
|
@ -363,179 +213,45 @@ private boolean validateToken(T identifier, byte[] password) {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if TokenInfo for the given identifier exists in database and if the
|
||||
* token is expired.
|
||||
*/
|
||||
public TokenInfo checkToken(T identifier) throws InvalidToken {
|
||||
TokenInfo info = currentTokens.get(identifier);
|
||||
if (info == null) {
|
||||
throw new InvalidToken("token " + formatTokenId(identifier)
|
||||
+ " can't be found in cache");
|
||||
}
|
||||
long now = Time.monotonicNow();
|
||||
if (info.getRenewDate() < now) {
|
||||
throw new InvalidToken("token " + formatTokenId(identifier) + " is " +
|
||||
"expired, current time: " + Time.formatTime(now) +
|
||||
" expected renewal time: " + Time.formatTime(info.getRenewDate()));
|
||||
}
|
||||
if (!validateToken(identifier, info.getPassword())) {
|
||||
throw new InvalidToken("Tampared/Inavalid token.");
|
||||
}
|
||||
return info;
|
||||
}
|
||||
|
||||
// TODO: handle roll private key/certificate
|
||||
private synchronized void removeExpiredKeys() {
|
||||
long now = Time.monotonicNow();
|
||||
for (Iterator<Map.Entry<Integer, OzoneSecretKey>> it = allKeys.entrySet()
|
||||
.iterator(); it.hasNext();) {
|
||||
Map.Entry<Integer, OzoneSecretKey> e = it.next();
|
||||
OzoneSecretKey key = e.getValue();
|
||||
if (key.getExpiryDate() < now && key.getExpiryDate() != -1) {
|
||||
if (!key.equals(currentKey)) {
|
||||
it.remove();
|
||||
try {
|
||||
store.removeTokenMasterKey(key);
|
||||
} catch (IOException ex) {
|
||||
LOG.error("Unable to remove master key " + key.getKeyId(), ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void loadTokenSecretState(OzoneManagerSecretState<T> state)
|
||||
throws IOException {
|
||||
LOG.info("Loading token state into token manager.");
|
||||
for (OzoneSecretKey key : state.ozoneManagerSecretState()) {
|
||||
allKeys.putIfAbsent(key.getKeyId(), key);
|
||||
}
|
||||
for (Map.Entry<T, Long> entry : state.getTokenState().entrySet()) {
|
||||
addPersistedDelegationToken(entry.getKey(), entry.getValue());
|
||||
}
|
||||
}
|
||||
|
||||
private String formatTokenId(T id) {
|
||||
public String formatTokenId(T id) {
|
||||
return "(" + id + ")";
|
||||
}
|
||||
|
||||
private void addPersistedDelegationToken(
|
||||
T identifier, long renewDate)
|
||||
throws IOException {
|
||||
if (running) {
|
||||
// a safety check
|
||||
throw new IOException(
|
||||
"Can't add persisted delegation token to a running SecretManager.");
|
||||
}
|
||||
int keyId = identifier.getMasterKeyId();
|
||||
OzoneSecretKey dKey = allKeys.get(keyId);
|
||||
if (dKey == null) {
|
||||
LOG.warn("No KEY found for persisted identifier "
|
||||
+ formatTokenId(identifier));
|
||||
return;
|
||||
}
|
||||
|
||||
PrivateKey privateKey = dKey.getPrivateKey();
|
||||
byte[] password = createPassword(identifier.getBytes(), privateKey);
|
||||
if (identifier.getSequenceNumber() > getDelegationTokenSeqNum()) {
|
||||
setDelegationTokenSeqNum(identifier.getSequenceNumber());
|
||||
}
|
||||
if (currentTokens.get(identifier) == null) {
|
||||
currentTokens.put(identifier, new TokenInfo(renewDate,
|
||||
password, identifier.getTrackingId()));
|
||||
} else {
|
||||
throw new IOException("Same delegation token being added twice: "
|
||||
+ formatTokenId(identifier));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Should be called before this object is used.
|
||||
*
|
||||
* @param keyPair
|
||||
* @throws IOException
|
||||
*/
|
||||
public void startThreads(KeyPair keyPair) throws IOException {
|
||||
Preconditions.checkState(!running);
|
||||
public synchronized void start(KeyPair keyPair) throws IOException {
|
||||
Preconditions.checkState(!isRunning());
|
||||
updateCurrentKey(keyPair);
|
||||
removeExpiredKeys();
|
||||
synchronized (this) {
|
||||
running = true;
|
||||
tokenRemoverThread = new Daemon(new ExpiredTokenRemover());
|
||||
tokenRemoverThread.start();
|
||||
}
|
||||
}
|
||||
|
||||
public void stopThreads() {
|
||||
if (LOG.isDebugEnabled()) {
|
||||
LOG.debug("Stopping expired delegation token remover thread");
|
||||
}
|
||||
running = false;
|
||||
|
||||
if (tokenRemoverThread != null) {
|
||||
synchronized (noInterruptsLock) {
|
||||
tokenRemoverThread.interrupt();
|
||||
}
|
||||
try {
|
||||
tokenRemoverThread.join();
|
||||
} catch (InterruptedException e) {
|
||||
throw new RuntimeException(
|
||||
"Unable to join on token removal thread", e);
|
||||
}
|
||||
}
|
||||
setIsRunning(true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stops the OzoneSecretManager.
|
||||
* Stops the OzoneDelegationTokenSecretManager.
|
||||
*
|
||||
* @throws IOException
|
||||
*/
|
||||
public void stop() throws IOException {
|
||||
stopThreads();
|
||||
if (this.store != null) {
|
||||
this.store.close();
|
||||
}
|
||||
public synchronized void stop() throws IOException {
|
||||
setIsRunning(false);
|
||||
}
|
||||
|
||||
/**
|
||||
* Update the current master key. This is called once by startThreads before
|
||||
* tokenRemoverThread is created,
|
||||
*/
|
||||
private void updateCurrentKey(KeyPair keyPair) throws IOException {
|
||||
LOG.info("Updating the current master key for generating tokens");
|
||||
|
||||
// TODO: fix me based on the certificate expire time to set the key
|
||||
// expire time.
|
||||
int newCurrentId = incrementCurrentKeyId();
|
||||
OzoneSecretKey newKey = new OzoneSecretKey(newCurrentId, -1,
|
||||
keyPair, maxKeyLength);
|
||||
|
||||
store.storeTokenMasterKey(newKey);
|
||||
if (!allKeys.containsKey(newKey.getKeyId())) {
|
||||
allKeys.put(newKey.getKeyId(), newKey);
|
||||
}
|
||||
|
||||
synchronized (this) {
|
||||
currentKey = newKey;
|
||||
}
|
||||
public String getDefaultSignatureAlgorithm() {
|
||||
return securityConfig.getSignatureAlgo();
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove expired delegation tokens from cache and persisted store.
|
||||
*/
|
||||
private void removeExpiredToken() throws IOException {
|
||||
long now = Time.monotonicNow();
|
||||
synchronized (this) {
|
||||
Iterator<Map.Entry<T,
|
||||
TokenInfo>> i = currentTokens.entrySet().iterator();
|
||||
while (i.hasNext()) {
|
||||
Map.Entry<T,
|
||||
TokenInfo> entry = i.next();
|
||||
long renewDate = entry.getValue().getRenewDate();
|
||||
if (renewDate < now) {
|
||||
i.remove();
|
||||
store.removeToken(entry.getKey());
|
||||
}
|
||||
}
|
||||
}
|
||||
public long getTokenMaxLifetime() {
|
||||
return tokenMaxLifetime;
|
||||
}
|
||||
|
||||
public long getTokenRenewInterval() {
|
||||
return tokenRenewInterval;
|
||||
}
|
||||
|
||||
public Text getService() {
|
||||
return service;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -547,52 +263,20 @@ public synchronized boolean isRunning() {
|
|||
return running;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns expiry time of a token given its identifier.
|
||||
*
|
||||
* @param dtId DelegationTokenIdentifier of a token
|
||||
* @return Expiry time of the token
|
||||
* @throws IOException
|
||||
*/
|
||||
public long getTokenExpiryTime(T dtId)
|
||||
throws IOException {
|
||||
TokenInfo info = currentTokens.get(dtId);
|
||||
if (info != null) {
|
||||
return info.getRenewDate();
|
||||
} else {
|
||||
throw new IOException("No delegation token found for this identifier");
|
||||
}
|
||||
public void setIsRunning(boolean val) {
|
||||
running = val;
|
||||
}
|
||||
|
||||
private class ExpiredTokenRemover extends Thread {
|
||||
private long lastTokenCacheCleanup;
|
||||
public OzoneSecretKey getCurrentKey() {
|
||||
return currentKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
LOG.info("Starting expired delegation token remover thread, "
|
||||
+ "tokenRemoverScanInterval=" + tokenRemoverScanInterval
|
||||
/ (60 * 1000) + " min(s)");
|
||||
try {
|
||||
while (running) {
|
||||
long now = Time.monotonicNow();
|
||||
if (lastTokenCacheCleanup + tokenRemoverScanInterval
|
||||
< now) {
|
||||
removeExpiredToken();
|
||||
lastTokenCacheCleanup = now;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(Math.min(5000,
|
||||
tokenRemoverScanInterval)); // 5 seconds
|
||||
} catch (InterruptedException ie) {
|
||||
LOG.error("ExpiredTokenRemover received " + ie);
|
||||
}
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
LOG.error("ExpiredTokenRemover thread received unexpected exception",
|
||||
t);
|
||||
Runtime.getRuntime().exit(-1);
|
||||
}
|
||||
}
|
||||
public AtomicInteger getCurrentKeyId() {
|
||||
return currentKeyId;
|
||||
}
|
||||
|
||||
public AtomicInteger getTokenSequenceNumber() {
|
||||
return tokenSequenceNumber;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,146 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.apache.hadoop.ozone.security;
|
||||
|
||||
import org.apache.hadoop.hdds.HddsConfigKeys;
|
||||
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
|
||||
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.BlockTokenSecretProto.AccessModeProto;
|
||||
import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
|
||||
import org.apache.hadoop.security.ssl.KeyStoreTestUtil;
|
||||
import org.apache.hadoop.security.token.Token;
|
||||
import org.apache.hadoop.test.GenericTestUtils;
|
||||
import org.apache.hadoop.test.LambdaTestUtils;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.junit.After;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.DataInputStream;
|
||||
import java.security.KeyPair;
|
||||
import java.security.Signature;
|
||||
import java.security.cert.X509Certificate;
|
||||
import java.util.EnumSet;
|
||||
|
||||
/**
|
||||
* Test class for {@link OzoneBlockTokenSecretManager}.
|
||||
*/
|
||||
public class TestOzoneBlockTokenSecretManager {
|
||||
|
||||
private OzoneBlockTokenSecretManager secretManager;
|
||||
private KeyPair keyPair;
|
||||
private X509Certificate x509Certificate;
|
||||
private long expiryTime;
|
||||
private String omCertSerialId;
|
||||
private static final String BASEDIR = GenericTestUtils
|
||||
.getTempPath(TestOzoneBlockTokenSecretManager.class.getSimpleName());
|
||||
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
OzoneConfiguration conf = new OzoneConfiguration();
|
||||
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, BASEDIR);
|
||||
// Create Ozone Master key pair.
|
||||
keyPair = KeyStoreTestUtil.generateKeyPair("RSA");
|
||||
expiryTime = Time.monotonicNow() + 60 * 60 * 24;
|
||||
// Create Ozone Master certificate (SCM CA issued cert) and key store.
|
||||
x509Certificate = KeyStoreTestUtil
|
||||
.generateCertificate("CN=OzoneMaster", keyPair, 30, "SHA256withRSA");
|
||||
omCertSerialId = x509Certificate.getSerialNumber().toString();
|
||||
secretManager = new OzoneBlockTokenSecretManager(conf,
|
||||
expiryTime, omCertSerialId);
|
||||
secretManager.start(keyPair);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {
|
||||
secretManager = null;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testGenerateToken() throws Exception {
|
||||
Token<OzoneBlockTokenIdentifier> token = secretManager.generateToken(
|
||||
"101", EnumSet.allOf(AccessModeProto.class), 100);
|
||||
OzoneBlockTokenIdentifier identifier =
|
||||
OzoneBlockTokenIdentifier.readFieldsProtobuf(new DataInputStream(
|
||||
new ByteArrayInputStream(token.getIdentifier())));
|
||||
// Check basic details.
|
||||
Assert.assertTrue(identifier.getBlockId().equals("101"));
|
||||
Assert.assertTrue(identifier.getAccessModes().equals(EnumSet
|
||||
.allOf(AccessModeProto.class)));
|
||||
Assert.assertTrue(identifier.getOmCertSerialId().equals(omCertSerialId));
|
||||
|
||||
validateHash(token.getPassword(), token.getIdentifier());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCreateIdentifierSuccess() throws Exception {
|
||||
OzoneBlockTokenIdentifier btIdentifier = secretManager.createIdentifier(
|
||||
"testUser", "101", EnumSet.allOf(AccessModeProto.class), 100);
|
||||
|
||||
// Check basic details.
|
||||
Assert.assertTrue(btIdentifier.getOwnerId().equals("testUser"));
|
||||
Assert.assertTrue(btIdentifier.getBlockId().equals("101"));
|
||||
Assert.assertTrue(btIdentifier.getAccessModes().equals(EnumSet
|
||||
.allOf(AccessModeProto.class)));
|
||||
Assert.assertTrue(btIdentifier.getOmCertSerialId().equals(omCertSerialId));
|
||||
|
||||
byte[] hash = secretManager.createPassword(btIdentifier);
|
||||
validateHash(hash, btIdentifier.getBytes());
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate hash using public key of KeyPair.
|
||||
* */
|
||||
private void validateHash(byte[] hash, byte[] identifier) throws Exception {
|
||||
Signature rsaSignature =
|
||||
Signature.getInstance(secretManager.getDefaultSignatureAlgorithm());
|
||||
rsaSignature.initVerify(keyPair.getPublic());
|
||||
rsaSignature.update(identifier);
|
||||
Assert.assertTrue(rsaSignature.verify(hash));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCreateIdentifierFailure() throws Exception {
|
||||
LambdaTestUtils.intercept(SecurityException.class,
|
||||
"Ozone block token can't be created without owner and access mode "
|
||||
+ "information.", () -> {
|
||||
secretManager.createIdentifier();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testRenewToken() throws Exception {
|
||||
LambdaTestUtils.intercept(UnsupportedOperationException.class,
|
||||
"Renew token operation is not supported for ozone block" +
|
||||
" tokens.", () -> {
|
||||
secretManager.renewToken(null, null);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCancelToken() throws Exception {
|
||||
LambdaTestUtils.intercept(UnsupportedOperationException.class,
|
||||
"Cancel token operation is not supported for ozone block" +
|
||||
" tokens.", () -> {
|
||||
secretManager.cancelToken(null, null);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -18,10 +18,6 @@
|
|||
|
||||
package org.apache.hadoop.ozone.security;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.security.KeyPair;
|
||||
import java.security.Signature;
|
||||
import org.apache.commons.io.FileUtils;
|
||||
import org.apache.hadoop.hdds.HddsConfigKeys;
|
||||
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
|
||||
|
|
@ -38,19 +34,25 @@
|
|||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
/**
|
||||
* Test class for {@link OzoneSecretManager}.
|
||||
*/
|
||||
public class TestOzoneSecretManager {
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.security.KeyPair;
|
||||
import java.security.Signature;
|
||||
|
||||
private OzoneSecretManager<OzoneTokenIdentifier> secretManager;
|
||||
/**
|
||||
* Test class for {@link OzoneDelegationTokenSecretManager}.
|
||||
*/
|
||||
public class TestOzoneDelegationTokenSecretManager {
|
||||
|
||||
private OzoneDelegationTokenSecretManager<OzoneTokenIdentifier>
|
||||
secretManager;
|
||||
private SecurityConfig securityConfig;
|
||||
private KeyPair keyPair;
|
||||
private long expiryTime;
|
||||
private Text serviceRpcAdd;
|
||||
private OzoneConfiguration conf;
|
||||
private static final String BASEDIR = GenericTestUtils
|
||||
.getTempPath(TestOzoneSecretManager.class.getSimpleName());
|
||||
private static final String BASEDIR = GenericTestUtils.getTempPath(
|
||||
TestOzoneDelegationTokenSecretManager.class.getSimpleName());
|
||||
private final static Text TEST_USER = new Text("testUser");
|
||||
private long tokenMaxLifetime = 1000 * 20;
|
||||
private long tokenRemoverScanInterval = 1000 * 20;
|
||||
|
|
@ -76,7 +78,7 @@ public void tearDown() throws IOException {
|
|||
public void testCreateToken() throws Exception {
|
||||
secretManager = createSecretManager(conf, tokenMaxLifetime,
|
||||
expiryTime, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -94,7 +96,7 @@ public void testCreateToken() throws Exception {
|
|||
public void testRenewTokenSuccess() throws Exception {
|
||||
secretManager = createSecretManager(conf, tokenMaxLifetime,
|
||||
expiryTime, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -110,7 +112,7 @@ public void testRenewTokenSuccess() throws Exception {
|
|||
public void testRenewTokenFailure() throws Exception {
|
||||
secretManager = createSecretManager(conf, tokenMaxLifetime,
|
||||
expiryTime, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -127,7 +129,7 @@ public void testRenewTokenFailure() throws Exception {
|
|||
public void testRenewTokenFailureMaxTime() throws Exception {
|
||||
secretManager = createSecretManager(conf, 100,
|
||||
100, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -145,7 +147,7 @@ public void testRenewTokenFailureMaxTime() throws Exception {
|
|||
public void testRenewTokenFailureRenewalTime() throws Exception {
|
||||
secretManager = createSecretManager(conf, 1000 * 10,
|
||||
10, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -159,7 +161,7 @@ public void testRenewTokenFailureRenewalTime() throws Exception {
|
|||
public void testCreateIdentifier() throws Exception {
|
||||
secretManager = createSecretManager(conf, tokenMaxLifetime,
|
||||
expiryTime, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
OzoneTokenIdentifier identifier = secretManager.createIdentifier();
|
||||
// Check basic details.
|
||||
Assert.assertTrue(identifier.getOwner().equals(new Text("")));
|
||||
|
|
@ -171,7 +173,7 @@ public void testCreateIdentifier() throws Exception {
|
|||
public void testCancelTokenSuccess() throws Exception {
|
||||
secretManager = createSecretManager(conf, tokenMaxLifetime,
|
||||
expiryTime, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -182,7 +184,7 @@ public void testCancelTokenSuccess() throws Exception {
|
|||
public void testCancelTokenFailure() throws Exception {
|
||||
secretManager = createSecretManager(conf, tokenMaxLifetime,
|
||||
expiryTime, tokenRemoverScanInterval);
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
Token<OzoneTokenIdentifier> token = secretManager.createToken(TEST_USER,
|
||||
TEST_USER,
|
||||
TEST_USER);
|
||||
|
|
@ -205,12 +207,12 @@ private void validateHash(byte[] hash, byte[] identifier) throws Exception {
|
|||
}
|
||||
|
||||
/**
|
||||
* Create instance of {@link OzoneSecretManager}.
|
||||
* Create instance of {@link OzoneDelegationTokenSecretManager}.
|
||||
*/
|
||||
private OzoneSecretManager<OzoneTokenIdentifier> createSecretManager(
|
||||
OzoneConfiguration config, long tokenMaxLife, long expiry, long
|
||||
tokenRemoverScanTime) throws IOException {
|
||||
return new OzoneSecretManager<>(config, tokenMaxLife,
|
||||
private OzoneDelegationTokenSecretManager<OzoneTokenIdentifier>
|
||||
createSecretManager(OzoneConfiguration config, long tokenMaxLife,
|
||||
long expiry, long tokenRemoverScanTime) throws IOException {
|
||||
return new OzoneDelegationTokenSecretManager<>(config, tokenMaxLife,
|
||||
expiry, tokenRemoverScanTime, serviceRpcAdd);
|
||||
}
|
||||
}
|
||||
|
|
@ -47,9 +47,9 @@
|
|||
import org.apache.hadoop.ipc.ProtobufRpcEngine;
|
||||
import org.apache.hadoop.ipc.RPC;
|
||||
import org.apache.hadoop.ozone.OzoneSecurityUtil;
|
||||
import org.apache.hadoop.ozone.security.OzoneSecretManager;
|
||||
import org.apache.hadoop.ozone.security.OzoneSecurityException;
|
||||
import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
|
||||
import org.apache.hadoop.ozone.security.OzoneSecretManager;
|
||||
import org.apache.hadoop.security.AccessControlException;
|
||||
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
|
||||
import org.apache.hadoop.metrics2.util.MBeans;
|
||||
|
|
@ -95,6 +95,7 @@
|
|||
import org.apache.hadoop.ozone.security.acl.OzoneObj.ResourceType;
|
||||
import org.apache.hadoop.ozone.security.acl.OzoneObjInfo;
|
||||
import org.apache.hadoop.ozone.security.acl.RequestContext;
|
||||
import org.apache.hadoop.ozone.security.OzoneDelegationTokenSecretManager;
|
||||
import org.apache.hadoop.security.SecurityUtil;
|
||||
import org.apache.hadoop.security.UserGroupInformation;
|
||||
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
|
||||
|
|
@ -111,6 +112,7 @@
|
|||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import javax.management.ObjectName;
|
||||
import javax.ws.rs.HEAD;
|
||||
import java.io.BufferedWriter;
|
||||
import java.io.File;
|
||||
import java.io.FileOutputStream;
|
||||
|
|
@ -128,7 +130,6 @@
|
|||
import java.util.Timer;
|
||||
import java.util.TimerTask;
|
||||
|
||||
import static org.apache.hadoop.ozone.security.OzoneSecurityException.ResultCodes.*;
|
||||
import static org.apache.hadoop.hdds.HddsUtils.getScmAddressForBlockClients;
|
||||
import static org.apache.hadoop.hdds.HddsUtils.getScmAddressForClients;
|
||||
import static org.apache.hadoop.hdds.HddsUtils.isHddsEnabled;
|
||||
|
|
@ -178,8 +179,8 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
|
|||
+ StartupOption.HELP.getName() + " ]\n";
|
||||
private static final String OM_DAEMON = "om";
|
||||
private static boolean securityEnabled = false;
|
||||
private static OzoneSecretManager secretManager;
|
||||
// TO DO: For testing purpose only, remove before commiting
|
||||
private static OzoneDelegationTokenSecretManager<OzoneTokenIdentifier>
|
||||
secretManager;
|
||||
private KeyPair keyPair;
|
||||
private CertificateClient certClient;
|
||||
private static boolean testSecureOmFlag = false;
|
||||
|
|
@ -367,9 +368,8 @@ private File getMetricsStorageFile() {
|
|||
}
|
||||
|
||||
|
||||
private OzoneSecretManager createSecretManager(
|
||||
OzoneConfiguration conf)
|
||||
throws IOException {
|
||||
private OzoneDelegationTokenSecretManager createSecretManager(
|
||||
OzoneConfiguration conf) throws IOException {
|
||||
long tokenRemoverScanInterval =
|
||||
conf.getTimeDuration(OMConfigKeys.DELEGATION_REMOVER_SCAN_INTERVAL_KEY,
|
||||
OMConfigKeys.DELEGATION_REMOVER_SCAN_INTERVAL_DEFAULT,
|
||||
|
|
@ -383,9 +383,8 @@ private OzoneSecretManager createSecretManager(
|
|||
OMConfigKeys.DELEGATION_TOKEN_RENEW_INTERVAL_DEFAULT,
|
||||
TimeUnit.MILLISECONDS);
|
||||
Text omRpcAddressTxt = new Text(OmUtils.getOmRpcAddress(configuration));
|
||||
|
||||
return new OzoneSecretManager(conf, tokenMaxLifetime, tokenRenewInterval,
|
||||
tokenRemoverScanInterval, omRpcAddressTxt);
|
||||
return new OzoneDelegationTokenSecretManager<>(conf, tokenMaxLifetime,
|
||||
tokenRenewInterval, tokenRemoverScanInterval, omRpcAddressTxt);
|
||||
}
|
||||
|
||||
private void stopSecretManager() throws IOException {
|
||||
|
|
@ -400,7 +399,7 @@ private void startSecretManager() {
|
|||
try {
|
||||
readKeyPair();
|
||||
LOG.info("Starting OM secret manager");
|
||||
secretManager.startThreads(keyPair);
|
||||
secretManager.start(keyPair);
|
||||
} catch (IOException e) {
|
||||
// Inability to start secret manager
|
||||
// can't be recovered from.
|
||||
|
|
@ -424,7 +423,8 @@ private void readKeyPair() throws OzoneSecurityException {
|
|||
certClient.getPrivateKey(OM_DAEMON));
|
||||
} catch (Exception e) {
|
||||
throw new OzoneSecurityException("Error reading private file for "
|
||||
+ "OzoneManager", e, OM_PUBLIC_PRIVATE_KEY_FILE_NOT_EXIST);
|
||||
+ "OzoneManager", e, OzoneSecurityException
|
||||
.ResultCodes.OM_PUBLIC_PRIVATE_KEY_FILE_NOT_EXIST);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,251 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.apache.hadoop.ozone.security;
|
||||
|
||||
import org.apache.commons.lang3.RandomStringUtils;
|
||||
import org.apache.commons.lang3.RandomUtils;
|
||||
import org.apache.hadoop.fs.FileUtil;
|
||||
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
|
||||
import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
|
||||
import org.apache.hadoop.security.ssl.KeyStoreTestUtil;
|
||||
import org.apache.hadoop.test.GenericTestUtils;
|
||||
import org.apache.hadoop.util.Time;
|
||||
import org.junit.After;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import javax.crypto.KeyGenerator;
|
||||
import javax.crypto.Mac;
|
||||
import javax.crypto.SecretKey;
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.security.GeneralSecurityException;
|
||||
import java.security.InvalidKeyException;
|
||||
import java.security.KeyPair;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.security.NoSuchProviderException;
|
||||
import java.security.PrivateKey;
|
||||
import java.security.Signature;
|
||||
import java.security.SignatureException;
|
||||
import java.security.cert.Certificate;
|
||||
import java.security.cert.CertificateEncodingException;
|
||||
import java.security.cert.X509Certificate;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.EnumSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Test class for OzoneManagerDelegationToken.
|
||||
*/
|
||||
public class TestOzoneManagerBlockToken {
|
||||
|
||||
private static final Logger LOG = LoggerFactory
|
||||
.getLogger(TestOzoneManagerBlockToken.class);
|
||||
private static final String BASEDIR = GenericTestUtils
|
||||
.getTempPath(TestOzoneManagerBlockToken.class.getSimpleName());
|
||||
private static final String KEYSTORES_DIR =
|
||||
new File(BASEDIR).getAbsolutePath();
|
||||
private static long expiryTime;
|
||||
private static KeyPair keyPair;
|
||||
private static X509Certificate cert;
|
||||
private static final long MAX_LEN = 1000;
|
||||
|
||||
@BeforeClass
|
||||
public static void setUp() throws Exception {
|
||||
File base = new File(BASEDIR);
|
||||
FileUtil.fullyDelete(base);
|
||||
base.mkdirs();
|
||||
expiryTime = Time.monotonicNow() + 60 * 60 * 24;
|
||||
|
||||
// Create Ozone Master key pair.
|
||||
keyPair = KeyStoreTestUtil.generateKeyPair("RSA");
|
||||
// Create Ozone Master certificate (SCM CA issued cert) and key store.
|
||||
cert = KeyStoreTestUtil
|
||||
.generateCertificate("CN=OzoneMaster", keyPair, 30, "SHA256withRSA");
|
||||
}
|
||||
|
||||
@After
|
||||
public void cleanUp() {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSignToken() throws GeneralSecurityException, IOException {
|
||||
String keystore = new File(KEYSTORES_DIR, "keystore.jks")
|
||||
.getAbsolutePath();
|
||||
String truststore = new File(KEYSTORES_DIR, "truststore.jks")
|
||||
.getAbsolutePath();
|
||||
String trustPassword = "trustPass";
|
||||
String keyStorePassword = "keyStorePass";
|
||||
String keyPassword = "keyPass";
|
||||
|
||||
|
||||
KeyStoreTestUtil.createKeyStore(keystore, keyStorePassword, keyPassword,
|
||||
"OzoneMaster", keyPair.getPrivate(), cert);
|
||||
|
||||
// Create trust store and put the certificate in the trust store
|
||||
Map<String, X509Certificate> certs = Collections.singletonMap("server",
|
||||
cert);
|
||||
KeyStoreTestUtil.createTrustStore(truststore, trustPassword, certs);
|
||||
|
||||
// Sign the OzoneMaster Token with Ozone Master private key
|
||||
PrivateKey privateKey = keyPair.getPrivate();
|
||||
OzoneBlockTokenIdentifier tokenId = new OzoneBlockTokenIdentifier(
|
||||
"testUser", "84940",
|
||||
EnumSet.allOf(HddsProtos.BlockTokenSecretProto.AccessModeProto.class),
|
||||
expiryTime, cert.getSerialNumber().toString(), MAX_LEN);
|
||||
byte[] signedToken = signTokenAsymmetric(tokenId, privateKey);
|
||||
|
||||
// Verify a valid signed OzoneMaster Token with Ozone Master
|
||||
// public key(certificate)
|
||||
boolean isValidToken = verifyTokenAsymmetric(tokenId, signedToken, cert);
|
||||
LOG.info("{} is {}", tokenId, isValidToken ? "valid." : "invalid.");
|
||||
|
||||
// Verify an invalid signed OzoneMaster Token with Ozone Master
|
||||
// public key(certificate)
|
||||
tokenId = new OzoneBlockTokenIdentifier("", "",
|
||||
EnumSet.allOf(HddsProtos.BlockTokenSecretProto.AccessModeProto.class),
|
||||
expiryTime, cert.getSerialNumber().toString(), MAX_LEN);
|
||||
LOG.info("Unsigned token {} is {}", tokenId,
|
||||
verifyTokenAsymmetric(tokenId, RandomUtils.nextBytes(128), cert));
|
||||
|
||||
}
|
||||
|
||||
public byte[] signTokenAsymmetric(OzoneBlockTokenIdentifier tokenId,
|
||||
PrivateKey privateKey) throws NoSuchAlgorithmException,
|
||||
InvalidKeyException, SignatureException {
|
||||
Signature rsaSignature = Signature.getInstance("SHA256withRSA");
|
||||
rsaSignature.initSign(privateKey);
|
||||
rsaSignature.update(tokenId.getBytes());
|
||||
byte[] signature = rsaSignature.sign();
|
||||
return signature;
|
||||
}
|
||||
|
||||
public boolean verifyTokenAsymmetric(OzoneBlockTokenIdentifier tokenId,
|
||||
byte[] signature, Certificate certificate) throws InvalidKeyException,
|
||||
NoSuchAlgorithmException, SignatureException {
|
||||
Signature rsaSignature = Signature.getInstance("SHA256withRSA");
|
||||
rsaSignature.initVerify(certificate);
|
||||
rsaSignature.update(tokenId.getBytes());
|
||||
boolean isValid = rsaSignature.verify(signature);
|
||||
return isValid;
|
||||
}
|
||||
|
||||
private byte[] signTokenSymmetric(OzoneBlockTokenIdentifier identifier,
|
||||
Mac mac, SecretKey key) {
|
||||
try {
|
||||
mac.init(key);
|
||||
} catch (InvalidKeyException ike) {
|
||||
throw new IllegalArgumentException("Invalid key to HMAC computation",
|
||||
ike);
|
||||
}
|
||||
return mac.doFinal(identifier.getBytes());
|
||||
}
|
||||
|
||||
OzoneBlockTokenIdentifier generateTestToken() {
|
||||
return new OzoneBlockTokenIdentifier(RandomStringUtils.randomAlphabetic(6),
|
||||
RandomStringUtils.randomAlphabetic(5),
|
||||
EnumSet.allOf(HddsProtos.BlockTokenSecretProto.AccessModeProto.class),
|
||||
expiryTime, cert.getSerialNumber().toString(), MAX_LEN);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAsymmetricTokenPerf() throws NoSuchAlgorithmException,
|
||||
CertificateEncodingException, NoSuchProviderException,
|
||||
InvalidKeyException, SignatureException {
|
||||
final int testTokenCount = 1000;
|
||||
List<OzoneBlockTokenIdentifier> tokenIds = new ArrayList<>();
|
||||
List<byte[]> tokenPasswordAsym = new ArrayList<>();
|
||||
for (int i = 0; i < testTokenCount; i++) {
|
||||
tokenIds.add(generateTestToken());
|
||||
}
|
||||
|
||||
KeyPair kp = KeyStoreTestUtil.generateKeyPair("RSA");
|
||||
|
||||
// Create Ozone Master certificate (SCM CA issued cert) and key store
|
||||
X509Certificate omCert;
|
||||
omCert = KeyStoreTestUtil.generateCertificate("CN=OzoneMaster",
|
||||
kp, 30, "SHA256withRSA");
|
||||
|
||||
long startTime = Time.monotonicNowNanos();
|
||||
for (int i = 0; i < testTokenCount; i++) {
|
||||
tokenPasswordAsym.add(
|
||||
signTokenAsymmetric(tokenIds.get(i), kp.getPrivate()));
|
||||
}
|
||||
long duration = Time.monotonicNowNanos() - startTime;
|
||||
LOG.info("Average token sign time with HmacSha256(RSA/1024 key) is {} ns",
|
||||
duration / testTokenCount);
|
||||
|
||||
startTime = Time.monotonicNowNanos();
|
||||
for (int i = 0; i < testTokenCount; i++) {
|
||||
verifyTokenAsymmetric(tokenIds.get(i), tokenPasswordAsym.get(i), omCert);
|
||||
}
|
||||
duration = Time.monotonicNowNanos() - startTime;
|
||||
LOG.info("Average token verify time with HmacSha256(RSA/1024 key) "
|
||||
+ "is {} ns", duration / testTokenCount);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSymmetricTokenPerf() {
|
||||
String hmacSHA1 = "HmacSHA1";
|
||||
String hmacSHA256 = "HmacSHA256";
|
||||
|
||||
testSymmetricTokenPerfHelper(hmacSHA1, 64);
|
||||
testSymmetricTokenPerfHelper(hmacSHA256, 1024);
|
||||
}
|
||||
|
||||
public void testSymmetricTokenPerfHelper(String hmacAlgorithm, int keyLen) {
|
||||
final int testTokenCount = 1000;
|
||||
List<OzoneBlockTokenIdentifier> tokenIds = new ArrayList<>();
|
||||
List<byte[]> tokenPasswordSym = new ArrayList<>();
|
||||
for (int i = 0; i < testTokenCount; i++) {
|
||||
tokenIds.add(generateTestToken());
|
||||
}
|
||||
|
||||
KeyGenerator keyGen;
|
||||
try {
|
||||
keyGen = KeyGenerator.getInstance(hmacAlgorithm);
|
||||
keyGen.init(keyLen);
|
||||
} catch (NoSuchAlgorithmException nsa) {
|
||||
throw new IllegalArgumentException("Can't find " + hmacAlgorithm +
|
||||
" algorithm.");
|
||||
}
|
||||
|
||||
Mac mac;
|
||||
try {
|
||||
mac = Mac.getInstance(hmacAlgorithm);
|
||||
} catch (NoSuchAlgorithmException nsa) {
|
||||
throw new IllegalArgumentException("Can't find " + hmacAlgorithm +
|
||||
" algorithm.");
|
||||
}
|
||||
|
||||
SecretKey secretKey = keyGen.generateKey();
|
||||
|
||||
long startTime = Time.monotonicNowNanos();
|
||||
for (int i = 0; i < testTokenCount; i++) {
|
||||
tokenPasswordSym.add(
|
||||
signTokenSymmetric(tokenIds.get(i), mac, secretKey));
|
||||
}
|
||||
long duration = Time.monotonicNowNanos() - startTime;
|
||||
LOG.info("Average token sign time with {}({} symmetric key) is {} ns",
|
||||
hmacAlgorithm, keyLen, duration / testTokenCount);
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue